This is an automated email from the ASF dual-hosted git repository. hsluoyz pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/casbin-gateway.git
commit f4e75057d389985f1730274683b7c7557521bfaa Author: Yang Luo <[email protected]> AuthorDate: Sun Sep 6 17:06:25 2026 +0800 feat(agents): drive an agent from Gateway and watch it answer --- agent/fingerprint.go | 5 + agent/fingerprints/claude-code.json | 24 ++ agent/fingerprints/codebuddy.json | 26 ++- agent/fingerprints/codex-cli.json | 24 +- agent/fingerprints/gemini-cli.json | 18 +- agent/fingerprints/qwen-code.json | 26 ++- agent/headless.go | 90 ++++++++ agentsession/command_unix.go | 45 ++++ agentsession/command_windows.go | 97 ++++++++ agentsession/event.go | 80 +++++++ agentsession/parse.go | 66 ++++++ agentsession/parse_claude.go | 175 ++++++++++++++ agentsession/parse_codex.go | 165 ++++++++++++++ agentsession/parse_text.go | 38 ++++ agentsession/runner.go | 202 +++++++++++++++++ agentsession/session.go | 100 ++++++++ agentsession/store.go | 361 +++++++++++++++++++++++++++++ controllers/agent_session.go | 297 ++++++++++++++++++++++++ main.go | 4 + object/agent_session.go | 149 ++++++++++++ object/ormer.go | 7 + routers/router.go | 9 + web/src/App.tsx | 2 + web/src/backend/DriveBackend.ts | 69 ++++++ web/src/locales/en/data.json | 15 ++ web/src/locales/zh/data.json | 15 ++ web/src/nav.ts | 4 + web/src/pages/ChatPage.tsx | 441 ++++++++++++++++++++++++++++++++++++ web/src/types.ts | 46 ++++ 29 files changed, 2596 insertions(+), 4 deletions(-) diff --git a/agent/fingerprint.go b/agent/fingerprint.go index 7fcded4..12cfe12 100644 --- a/agent/fingerprint.go +++ b/agent/fingerprint.go @@ -103,6 +103,11 @@ type Fingerprint struct { // both installs and upgrades. It runs through the platform shell, so it is // shown in full before anyone clicks it. InstallScript *InstallScript `json:"installScript,omitempty"` + + // Headless is the agent's own non-interactive mode, which is what Gateway + // drives it through. Nil for an agent that publishes none: it can still be + // watched and configured, just not driven. + Headless *Headless `json:"headless,omitempty"` } // InstallScript is the vendor's installer command per platform. An empty diff --git a/agent/fingerprints/claude-code.json b/agent/fingerprints/claude-code.json index 34cd951..5e5232a 100644 --- a/agent/fingerprints/claude-code.json +++ b/agent/fingerprints/claude-code.json @@ -23,5 +23,29 @@ "windows": "irm https://claude.ai/install.ps1 | iex", "darwin": "curl -fsSL https://claude.ai/install.sh | bash", "linux": "curl -fsSL https://claude.ai/install.sh | bash" + }, + "headless": { + "args": [ + "-p", + "--output-format", + "stream-json", + "--verbose", + "--session-id", + "{session}" + ], + "resumeArgs": [ + "-p", + "--output-format", + "stream-json", + "--verbose", + "--resume", + "{session}" + ], + "modelArgs": [ + "--model", + "{model}" + ], + "format": "claude-stream-json", + "promptStdin": true } } diff --git a/agent/fingerprints/codebuddy.json b/agent/fingerprints/codebuddy.json index 48949f9..aecf7f5 100644 --- a/agent/fingerprints/codebuddy.json +++ b/agent/fingerprints/codebuddy.json @@ -5,5 +5,29 @@ "execName": "codebuddy", "instanceEnv": "CODEBUDDY_CONFIG_DIR", "stateDir": "codebuddy", - "npmPackage": "@tencent-ai/codebuddy-code" + "npmPackage": "@tencent-ai/codebuddy-code", + "headless": { + "args": [ + "-p", + "--output-format", + "stream-json", + "--verbose", + "--session-id", + "{session}" + ], + "resumeArgs": [ + "-p", + "--output-format", + "stream-json", + "--verbose", + "--resume", + "{session}" + ], + "modelArgs": [ + "--model", + "{model}" + ], + "format": "claude-stream-json", + "promptStdin": true + } } diff --git a/agent/fingerprints/codex-cli.json b/agent/fingerprints/codex-cli.json index 8f3a755..58302f9 100644 --- a/agent/fingerprints/codex-cli.json +++ b/agent/fingerprints/codex-cli.json @@ -16,5 +16,27 @@ ], "updateArgs": [ "update" - ] + ], + "headless": { + "args": [ + "exec", + "--json", + "--skip-git-repo-check", + "-" + ], + "resumeArgs": [ + "exec", + "resume", + "--json", + "--skip-git-repo-check", + "{session}", + "-" + ], + "modelArgs": [ + "--model", + "{model}" + ], + "format": "codex-json", + "promptStdin": true + } } diff --git a/agent/fingerprints/gemini-cli.json b/agent/fingerprints/gemini-cli.json index 1d3820e..be4251c 100644 --- a/agent/fingerprints/gemini-cli.json +++ b/agent/fingerprints/gemini-cli.json @@ -4,5 +4,21 @@ "installUrl": "https://google-gemini.github.io/gemini-cli/";, "execName": "gemini", "stateDir": "gemini", - "npmPackage": "@google/gemini-cli" + "npmPackage": "@google/gemini-cli", + "headless": { + "args": [ + "-p", + "", + "-o", + "text", + "--session-id", + "{session}" + ], + "modelArgs": [ + "-m", + "{model}" + ], + "format": "text", + "promptStdin": true + } } diff --git a/agent/fingerprints/qwen-code.json b/agent/fingerprints/qwen-code.json index 9e4b838..43b94f4 100644 --- a/agent/fingerprints/qwen-code.json +++ b/agent/fingerprints/qwen-code.json @@ -4,5 +4,29 @@ "installUrl": "https://github.com/QwenLM/qwen-code";, "execName": "qwen", "stateDir": "qwen", - "npmPackage": "@qwen-code/qwen-code" + "npmPackage": "@qwen-code/qwen-code", + "headless": { + "args": [ + "-p", + "", + "-o", + "text", + "--session-id", + "{session}" + ], + "resumeArgs": [ + "-p", + "", + "-o", + "text", + "--resume", + "{session}" + ], + "modelArgs": [ + "-m", + "{model}" + ], + "format": "text", + "promptStdin": true + } } diff --git a/agent/headless.go b/agent/headless.go new file mode 100644 index 0000000..2fee7a0 --- /dev/null +++ b/agent/headless.go @@ -0,0 +1,90 @@ +// Copyright 2026 The casbin Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package agent + +// Placeholders one of Headless's command lines carries. Each is replaced whole: +// a placeholder is always an argument of its own, never part of one, so nothing +// here has to be quoted or escaped. +const ( + PromptPlaceholder = "{prompt}" + SessionPlaceholder = "{session}" + ModelPlaceholder = "{model}" +) + +// Headless is how one prompt is handed to an agent without a terminal, which is +// what lets Gateway drive it on somebody's behalf. +// +// Args and ResumeArgs are whole command lines rather than a base and an +// addition: the agents disagree about where a resumed prompt goes - Codex takes +// a subcommand, the Claude Code family a flag - and writing both out is what +// keeps that disagreement out of the code. +type Headless struct { + // Args starts a new conversation. An agent that carries SessionPlaceholder + // here is told which session it is starting, so Gateway names it rather than + // reading the name back out of the output. + Args []string `json:"args"` + // ResumeArgs carries on the conversation Gateway last started. Empty for an + // agent that cannot resume one, whose every turn then stands alone. + ResumeArgs []string `json:"resumeArgs,omitempty"` + // ModelArgs are appended when a model was picked for the session, and left + // out entirely when none was, so the agent keeps its own default. + ModelArgs []string `json:"modelArgs,omitempty"` + // Format names the parser for what the agent writes to stdout. + Format string `json:"format"` + // PromptStdin hands the prompt on stdin instead of the command line. It is + // what every agent driven from a chat window needs: a package manager's + // Windows shim runs through cmd.exe, which ends an argument at the first + // newline, and a typed message routinely has one. + PromptStdin bool `json:"promptStdin,omitempty"` +} + +// CanResume reports whether the agent carries a conversation across turns. +func (h *Headless) CanResume() bool { + return len(h.ResumeArgs) > 0 +} + +// NamesSession reports whether the agent is told the id of the session it is +// starting. One that is not leaves the id to be read out of its output. +func (h *Headless) NamesSession() bool { + for _, arg := range h.Args { + if arg == SessionPlaceholder { + return true + } + } + return false +} + +// HeadlessOf is how one agent is driven without a terminal, nil for an agent +// that publishes no such mode. It reads the fingerprints rather than a host +// scan, so the answer is known while the agent is not installed. +func HeadlessOf(agentId string) *Headless { + for i := range fingerprints { + if fingerprints[i].ID == agentId { + return fingerprints[i].Headless + } + } + return nil +} + +// DrivableAgents are the ids of every agent Gateway can drive. +func DrivableAgents() []string { + result := []string{} + for i := range fingerprints { + if fingerprints[i].Headless != nil { + result = append(result, fingerprints[i].ID) + } + } + return result +} diff --git a/agentsession/command_unix.go b/agentsession/command_unix.go new file mode 100644 index 0000000..aee4490 --- /dev/null +++ b/agentsession/command_unix.go @@ -0,0 +1,45 @@ +// Copyright 2026 The casbin Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//go:build !windows + +package agentsession + +import ( + "context" + "os/exec" + "syscall" +) + +// newCommand runs the launcher in a process group of its own, which is what +// lets everything it starts be stopped together. +func newCommand(ctx context.Context, executable string, args []string) *exec.Cmd { + cmd := exec.Command(executable, args...) + cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} + return cmd +} + +// killTree ends the launcher and everything it started. Every one of these +// agents is a shim in front of a runtime, so killing the shim alone would leave +// the agent running. +func killTree(cmd *exec.Cmd) { + if cmd.Process == nil { + return + } + if pgid, err := syscall.Getpgid(cmd.Process.Pid); err == nil { + syscall.Kill(-pgid, syscall.SIGKILL) + return + } + cmd.Process.Kill() +} diff --git a/agentsession/command_windows.go b/agentsession/command_windows.go new file mode 100644 index 0000000..55b2a59 --- /dev/null +++ b/agentsession/command_windows.go @@ -0,0 +1,97 @@ +// Copyright 2026 The casbin Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//go:build windows + +package agentsession + +import ( + "context" + "os/exec" + "path/filepath" + "strconv" + "strings" + "syscall" +) + +// newCommand runs the launcher with its output on a pipe and no console of its +// own. A package manager's shim is a batch file, which Windows cannot start +// directly, so those go through cmd.exe. +func newCommand(ctx context.Context, executable string, args []string) *exec.Cmd { + switch strings.ToLower(filepath.Ext(executable)) { + case ".cmd", ".bat": + cmd := exec.Command("cmd.exe") + cmd.SysProcAttr = &syscall.SysProcAttr{HideWindow: true, CmdLine: batchLine(executable, args)} + return cmd + default: + cmd := exec.Command(executable, args...) + cmd.SysProcAttr = &syscall.SysProcAttr{HideWindow: true} + return cmd + } +} + +// batchLine writes the command line for a shim run through cmd.exe, which parses +// what it is given a second time: os/exec quotes for the program's own runtime, +// and an argument that cmd.exe would read as punctuation - "&" and its like - +// survives that only inside quotes. So every argument is quoted here, whether it +// needs it or not, and the whole command is wrapped for "/s", which takes what +// is between the outer quotes literally. +func batchLine(executable string, args []string) string { + line := quoteForCmd(executable) + for _, arg := range args { + line += " " + quoteForCmd(arg) + } + return `cmd.exe /s /c "` + line + `"` +} + +// quoteForCmd wraps one argument in quotes, escaping the quotes and backslashes +// inside it the way a program's own runtime unpicks them. +func quoteForCmd(arg string) string { + quoted := strings.Builder{} + quoted.WriteByte('"') + + backslashes := 0 + for i := 0; i < len(arg); i++ { + switch character := arg[i]; character { + case '\\': + backslashes++ + case '"': + // The backslashes before a quote are doubled, then the quote itself + // is escaped with one more. + quoted.WriteString(strings.Repeat(`\`, backslashes*2+1)) + quoted.WriteByte('"') + backslashes = 0 + default: + quoted.WriteString(strings.Repeat(`\`, backslashes)) + quoted.WriteByte(character) + backslashes = 0 + } + } + // Trailing backslashes would otherwise escape the closing quote. + quoted.WriteString(strings.Repeat(`\`, backslashes*2)) + quoted.WriteByte('"') + return quoted.String() +} + +// killTree ends the launcher and everything it started. Every one of these +// agents is a shim in front of a runtime, so killing the shim alone would leave +// the agent running. +func killTree(cmd *exec.Cmd) { + if cmd.Process == nil { + return + } + kill := exec.Command("taskkill", "/F", "/T", "/PID", strconv.Itoa(cmd.Process.Pid)) + kill.SysProcAttr = &syscall.SysProcAttr{HideWindow: true} + kill.Run() +} diff --git a/agentsession/event.go b/agentsession/event.go new file mode 100644 index 0000000..57a663e --- /dev/null +++ b/agentsession/event.go @@ -0,0 +1,80 @@ +// Copyright 2026 The casbin Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package agentsession + +import "time" + +// EventType is what one piece of an agent's answer is. +type EventType string + +const ( + // EventPrompt is what was asked. It is on the same feed as the answer so a + // page that reconnects reads the whole conversation back from one place. + EventPrompt EventType = "prompt" + EventText EventType = "text" + EventThinking EventType = "thinking" + EventToolUse EventType = "toolUse" + EventToolResult EventType = "toolResult" + EventUsage EventType = "usage" + EventDone EventType = "done" + EventError EventType = "error" +) + +// Event is one piece of what an agent said, in a shape every agent is flattened +// into. The parsers own the differences so nothing downstream - the page, the +// chat bridges, the records - has to know which agent produced it. +type Event struct { + Type EventType `json:"type"` + // Seq counts the events of one session from the first. A page that lost its + // connection says which one it had, and is sent only what came after, so a + // reconnect does not repeat the conversation. + Seq int64 `json:"seq"` + CreatedTime string `json:"createdTime"` + + // Text is the answer for EventText and EventThinking, and the message for + // EventError. + Text string `json:"text,omitempty"` + + ToolName string `json:"toolName,omitempty"` + ToolUseId string `json:"toolUseId,omitempty"` + // ToolInput is the arguments the agent called the tool with, as JSON. It is + // what an approval prompt has to show for a decision to mean anything. + ToolInput string `json:"toolInput,omitempty"` + + Model string `json:"model,omitempty"` + // NativeId is the agent's own id for this conversation, which is what a + // later turn resumes. Reported by the parser for an agent Gateway cannot + // name a session for up front. + NativeId string `json:"nativeId,omitempty"` + + InputTokens int `json:"inputTokens,omitempty"` + OutputTokens int `json:"outputTokens,omitempty"` + CostUsd float64 `json:"costUsd,omitempty"` + DurationMs int64 `json:"durationMs,omitempty"` +} + +func newEvent(eventType EventType) Event { + return Event{Type: eventType, CreatedTime: time.Now().Format(time.RFC3339Nano)} +} + +func textEvent(eventType EventType, text string) Event { + event := newEvent(eventType) + event.Text = text + return event +} + +func errorEvent(err error) Event { + return textEvent(EventError, err.Error()) +} diff --git a/agentsession/parse.go b/agentsession/parse.go new file mode 100644 index 0000000..2948939 --- /dev/null +++ b/agentsession/parse.go @@ -0,0 +1,66 @@ +// Copyright 2026 The casbin Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package agentsession + +import ( + "bufio" + "encoding/json" + "io" +) + +// The parsers, named by a fingerprint's "format". +const ( + // FormatClaudeStreamJson is "--output-format stream-json", which the Claude + // Code family all speak. + FormatClaudeStreamJson = "claude-stream-json" + // FormatCodexJson is "codex exec --json". + FormatCodexJson = "codex-json" + // FormatText is whatever the agent prints, taken as the answer. It is the + // fallback for an agent with no machine-readable mode, and loses the tool + // calls: those are still recorded by the monitor, just not shown here. + FormatText = "text" +) + +// maxLine is the longest line a parser accepts. One line can carry a whole file +// an agent just read, so the default scanner limit is far too small. +const maxLine = 8 * 1024 * 1024 + +func parse(format string, reader io.Reader, emit func(Event)) error { + scanner := bufio.NewScanner(reader) + scanner.Buffer(make([]byte, 0, 64*1024), maxLine) + + switch format { + case FormatClaudeStreamJson: + return parseClaude(scanner, emit) + case FormatCodexJson: + return parseCodex(scanner, emit) + default: + return parseText(scanner, emit) + } +} + +// jsonString renders a value that an agent may write either as a string or as +// something structured, which is how tool arguments and tool results arrive. +func jsonString(raw json.RawMessage) string { + if len(raw) == 0 { + return "" + } + + text := "" + if err := json.Unmarshal(raw, &text); err == nil { + return text + } + return string(raw) +} diff --git a/agentsession/parse_claude.go b/agentsession/parse_claude.go new file mode 100644 index 0000000..974ffab --- /dev/null +++ b/agentsession/parse_claude.go @@ -0,0 +1,175 @@ +// Copyright 2026 The casbin Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package agentsession + +import ( + "bufio" + "encoding/json" + "errors" + "strings" +) + +// claudeLine is one record of "--output-format stream-json". Only the fields +// Gateway shows are named; the rest are ignored, which is what keeps this +// working across the releases that add to them. +type claudeLine struct { + Type string `json:"type"` + Subtype string `json:"subtype"` + SessionId string `json:"session_id"` + Model string `json:"model"` + + Message *claudeMessage `json:"message"` + + // The closing record of a run. + IsError bool `json:"is_error"` + Result string `json:"result"` + TotalCostUsd float64 `json:"total_cost_usd"` + DurationMs int64 `json:"duration_ms"` + Usage *claudeUsage `json:"usage"` +} + +type claudeMessage struct { + Model string `json:"model"` + Content json.RawMessage `json:"content"` + Usage *claudeUsage `json:"usage"` +} + +type claudeBlock struct { + Type string `json:"type"` + + Text string `json:"text"` + Thinking string `json:"thinking"` + + Id string `json:"id"` + Name string `json:"name"` + Input json.RawMessage `json:"input"` + + ToolUseId string `json:"tool_use_id"` + Content json.RawMessage `json:"content"` +} + +type claudeUsage struct { + InputTokens int `json:"input_tokens"` + OutputTokens int `json:"output_tokens"` +} + +func parseClaude(scanner *bufio.Scanner, emit func(Event)) error { + failure := "" + // The agent names its session on every record it writes, and says so several + // times before it answers. It is reported once. + named := false + for scanner.Scan() { + line := strings.TrimSpace(scanner.Text()) + if line == "" || line[0] != '{' { + continue + } + + record := claudeLine{} + if err := json.Unmarshal([]byte(line), &record); err != nil { + continue + } + + if !named && record.SessionId != "" { + named = true + event := newEvent(EventUsage) + event.NativeId = record.SessionId + event.Model = record.Model + emit(event) + } + + switch record.Type { + case "assistant", "user": + emitClaudeBlocks(record, emit) + case "result": + failure = emitClaudeResult(record, emit) + } + } + + if err := scanner.Err(); err != nil { + return err + } + if failure != "" { + return errors.New(failure) + } + return nil +} + +func emitClaudeBlocks(record claudeLine, emit func(Event)) { + if record.Message == nil { + return + } + + blocks := []claudeBlock{} + if err := json.Unmarshal(record.Message.Content, &blocks); err != nil { + // A message whose content is a bare string is the whole answer. + if text := jsonString(record.Message.Content); text != "" { + emit(textEvent(EventText, text)) + } + return + } + + for _, block := range blocks { + switch block.Type { + case "text": + if block.Text != "" { + event := textEvent(EventText, block.Text) + event.Model = record.Message.Model + emit(event) + } + case "thinking": + if block.Thinking != "" { + emit(textEvent(EventThinking, block.Thinking)) + } + case "tool_use": + event := newEvent(EventToolUse) + event.ToolName = block.Name + event.ToolUseId = block.Id + event.ToolInput = jsonString(block.Input) + emit(event) + case "tool_result": + event := newEvent(EventToolResult) + event.ToolUseId = block.ToolUseId + event.Text = jsonString(block.Content) + emit(event) + } + } +} + +// emitClaudeResult reports the closing record and returns why the run failed, +// empty when it did not. +func emitClaudeResult(record claudeLine, emit func(Event)) string { + event := newEvent(EventUsage) + event.CostUsd = record.TotalCostUsd + event.DurationMs = record.DurationMs + if record.Usage != nil { + event.InputTokens = record.Usage.InputTokens + event.OutputTokens = record.Usage.OutputTokens + } + // A run that reported nothing it cost says nothing here either. + if event.CostUsd > 0 || event.DurationMs > 0 || event.InputTokens > 0 || event.OutputTokens > 0 { + emit(event) + } + + if !record.IsError { + return "" + } + if record.Result != "" { + return record.Result + } + if record.Subtype != "" { + return record.Subtype + } + return "the agent ended with an error" +} diff --git a/agentsession/parse_codex.go b/agentsession/parse_codex.go new file mode 100644 index 0000000..4e2ec37 --- /dev/null +++ b/agentsession/parse_codex.go @@ -0,0 +1,165 @@ +// Copyright 2026 The casbin Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package agentsession + +import ( + "bufio" + "encoding/json" + "errors" + "strings" +) + +// codexLine is one record of "codex exec --json". Codex has written two shapes +// of these: the thread-and-item one, and an older one that wrapped everything +// in "msg". Both are read, because which one an installation speaks depends on +// its version rather than on anything Gateway controls. +type codexLine struct { + Type string `json:"type"` + ThreadId string `json:"thread_id"` + Item *codexItem `json:"item"` + Usage *codexUsage `json:"usage"` + Error *codexError `json:"error"` + Message string `json:"message"` + + Msg *codexItem `json:"msg"` +} + +type codexItem struct { + Type string `json:"type"` + ItemType string `json:"item_type"` + + Text string `json:"text"` + Message string `json:"message"` + + Command string `json:"command"` + AggregatedOutput string `json:"aggregated_output"` + + Server string `json:"server"` + Tool string `json:"tool"` + Args json.RawMessage `json:"arguments"` +} + +type codexUsage struct { + InputTokens int `json:"input_tokens"` + OutputTokens int `json:"output_tokens"` +} + +type codexError struct { + Message string `json:"message"` +} + +func parseCodex(scanner *bufio.Scanner, emit func(Event)) error { + failure := "" + for scanner.Scan() { + line := strings.TrimSpace(scanner.Text()) + if line == "" || line[0] != '{' { + continue + } + + record := codexLine{} + if err := json.Unmarshal([]byte(line), &record); err != nil { + continue + } + + switch { + case record.Type == "thread.started" && record.ThreadId != "": + event := newEvent(EventUsage) + event.NativeId = record.ThreadId + emit(event) + // Only a completed item is reported: the started and updated ones carry + // the same text again as it grows, and showing each would repeat it. + case record.Type == "item.completed" && record.Item != nil: + emitCodexItem(record.Item, emit) + case record.Type == "turn.completed": + if record.Usage != nil { + event := newEvent(EventUsage) + event.InputTokens = record.Usage.InputTokens + event.OutputTokens = record.Usage.OutputTokens + emit(event) + } + case record.Type == "turn.failed" || record.Type == "error": + failure = codexFailure(record) + case record.Msg != nil: + emitCodexItem(record.Msg, emit) + } + } + + if err := scanner.Err(); err != nil { + return err + } + if failure != "" { + return errors.New(failure) + } + return nil +} + +func emitCodexItem(item *codexItem, emit func(Event)) { + itemType := item.ItemType + if itemType == "" { + itemType = item.Type + } + + switch itemType { + case "agent_message": + if text := firstOf(item.Text, item.Message); text != "" { + emit(textEvent(EventText, text)) + } + case "reasoning", "agent_reasoning": + if text := firstOf(item.Text, item.Message); text != "" { + emit(textEvent(EventThinking, text)) + } + case "command_execution", "exec_command_begin": + event := newEvent(EventToolUse) + event.ToolName = "Shell" + event.ToolInput = item.Command + emit(event) + if item.AggregatedOutput != "" { + emit(textEvent(EventToolResult, item.AggregatedOutput)) + } + case "mcp_tool_call": + event := newEvent(EventToolUse) + event.ToolName = strings.TrimPrefix(item.Server+"."+item.Tool, ".") + event.ToolInput = jsonString(item.Args) + emit(event) + case "file_change", "patch_apply_begin": + event := newEvent(EventToolUse) + event.ToolName = "Edit" + event.ToolInput = firstOf(item.Text, item.Message) + emit(event) + case "error": + if text := firstOf(item.Message, item.Text); text != "" { + emit(textEvent(EventError, text)) + } + } +} + +func codexFailure(record codexLine) string { + if record.Error != nil && record.Error.Message != "" { + return record.Error.Message + } + if record.Message != "" { + return record.Message + } + return "the agent ended with an error" +} + +func firstOf(values ...string) string { + for _, value := range values { + if value != "" { + return value + } + } + return "" +} diff --git a/agentsession/parse_text.go b/agentsession/parse_text.go new file mode 100644 index 0000000..76482aa --- /dev/null +++ b/agentsession/parse_text.go @@ -0,0 +1,38 @@ +// Copyright 2026 The casbin Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package agentsession + +import ( + "bufio" + "strings" +) + +// parseText takes whatever the agent printed as the answer. Lines are sent on +// as they arrive rather than gathered up, so a long answer appears while it is +// being written instead of all at once at the end. +func parseText(scanner *bufio.Scanner, emit func(Event)) error { + said := false + for scanner.Scan() { + line := scanner.Text() + // The leading blank lines are the agent clearing its own progress + // output, and are not part of what it said. + if !said && strings.TrimSpace(line) == "" { + continue + } + said = true + emit(textEvent(EventText, line+"\n")) + } + return scanner.Err() +} diff --git a/agentsession/runner.go b/agentsession/runner.go new file mode 100644 index 0000000..9eb4ea2 --- /dev/null +++ b/agentsession/runner.go @@ -0,0 +1,202 @@ +// Copyright 2026 The casbin Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package agentsession + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "os" + "strings" + "time" + + "github.com/apache/casbin-gateway/agent" + "github.com/apache/casbin-gateway/agenthome" +) + +// turnTimeout caps one exchange. An agent driven from a chat window has nobody +// watching the console it would otherwise hang in, so a turn that goes quiet is +// ended rather than left holding the session forever. +const turnTimeout = 30 * time.Minute + +// stderrLimit is how much of a failed run's diagnostics is kept for the error +// message. Agents are talkative on the way down and only the tail is useful. +const stderrLimit = 4000 + +// run carries out one exchange and reports what went wrong, if anything. +func run(ctx context.Context, headless *agent.Headless, session Session, prompt string, emit func(Event)) error { + launch := agent.LaunchOf(agent.Installation{ + AgentId: session.AgentId, + Path: session.AgentPath, + Owner: session.Owner, + }) + if launch.Executable == "" { + return errors.New("no launcher was found for this installation") + } + + // A command line cannot carry a newline through a package manager's Windows + // shim - cmd.exe ends the argument there - and a message typed in a chat + // window routinely has one. An agent whose prompt goes on the command line + // is therefore held to a single line rather than quietly losing the rest. + if !headless.PromptStdin && strings.ContainsAny(prompt, "\r\n") { + return errors.New("this agent takes its prompt on the command line, which cannot carry more than one line") + } + + workDir, err := resolveWorkDir(session.WorkDir) + if err != nil { + return err + } + + ctx, cancel := context.WithTimeout(ctx, turnTimeout) + defer cancel() + + args := buildArgs(headless, session, prompt) + cmd := newCommand(ctx, launch.Executable, args) + cmd.Dir = workDir + cmd.Env = os.Environ() + + stdout, err := cmd.StdoutPipe() + if err != nil { + return err + } + stderr := &boundedBuffer{limit: stderrLimit} + cmd.Stderr = stderr + + // The prompt goes on stdin wherever the agent reads it there: it is the one + // part of the command line that carries arbitrary text, and keeping it off + // removes every quoting question with it. Stdin is closed either way, so an + // agent that decides to ask something gets an answer instead of waiting. + stdin, err := cmd.StdinPipe() + if err != nil { + return err + } + go func() { + defer stdin.Close() + if headless.PromptStdin { + io.WriteString(stdin, prompt) + } + }() + + if err := cmd.Start(); err != nil { + return err + } + + // Killing the process itself leaves what it started behind - every one of + // these agents is a launcher in front of a runtime - so the tree goes. + stopped := make(chan struct{}) + go func() { + select { + case <-ctx.Done(): + killTree(cmd) + case <-stopped: + } + }() + + parseErr := parse(headless.Format, stdout, emit) + waitErr := cmd.Wait() + close(stopped) + + if ctxErr := ctx.Err(); ctxErr != nil { + if errors.Is(ctxErr, context.DeadlineExceeded) { + return fmt.Errorf("the agent was still running after %s and was stopped", turnTimeout) + } + return errors.New("stopped") + } + if waitErr != nil { + if detail := strings.TrimSpace(stderr.String()); detail != "" { + return fmt.Errorf("%s: %s", waitErr, detail) + } + return waitErr + } + return parseErr +} + +// buildArgs writes the command line for one turn. A session the agent can carry +// on is resumed; one it cannot starts over, which is the honest behaviour for an +// agent that publishes no way to continue. +func buildArgs(headless *agent.Headless, session Session, prompt string) []string { + template := headless.Args + if session.NativeId != "" && session.Turns > 0 && headless.CanResume() { + template = headless.ResumeArgs + } + + args := fill(template, session, prompt, headless.PromptStdin) + if session.Model != "" && len(headless.ModelArgs) > 0 { + args = append(args, fill(headless.ModelArgs, session, prompt, headless.PromptStdin)...) + } + return args +} + +// fill replaces the placeholders, each of which is a whole argument. A prompt +// handed over on stdin drops out of the command line rather than being repeated +// on it. +func fill(template []string, session Session, prompt string, promptOnStdin bool) []string { + args := make([]string, 0, len(template)) + for _, arg := range template { + switch arg { + case agent.PromptPlaceholder: + if !promptOnStdin { + args = append(args, prompt) + } + case agent.SessionPlaceholder: + args = append(args, session.NativeId) + case agent.ModelPlaceholder: + args = append(args, session.Model) + default: + args = append(args, arg) + } + } + return args +} + +// resolveWorkDir is the directory the agent runs in. An empty one is the home of +// the account Gateway runs as, which is where an agent started by hand would +// have been. +func resolveWorkDir(workDir string) (string, error) { + if workDir == "" { + return agenthome.Resolve("") + } + + info, err := os.Stat(workDir) + if err != nil { + return "", fmt.Errorf("the working directory cannot be opened: %w", err) + } + if !info.IsDir() { + return "", fmt.Errorf("%s is not a directory", workDir) + } + return workDir, nil +} + +// boundedBuffer keeps the tail of what was written to it. +type boundedBuffer struct { + limit int + buf bytes.Buffer +} + +func (b *boundedBuffer) Write(p []byte) (int, error) { + written, err := b.buf.Write(p) + if b.buf.Len() > b.limit { + kept := b.buf.Bytes()[b.buf.Len()-b.limit:] + trimmed := bytes.NewBuffer(append([]byte{}, kept...)) + b.buf = *trimmed + } + return written, err +} + +func (b *boundedBuffer) String() string { + return b.buf.String() +} diff --git a/agentsession/session.go b/agentsession/session.go new file mode 100644 index 0000000..8c53763 --- /dev/null +++ b/agentsession/session.go @@ -0,0 +1,100 @@ +// Copyright 2026 The casbin Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package agentsession drives an installed agent on somebody's behalf: it hands +// one prompt to the agent's own non-interactive mode and flattens what comes +// back into events. Nothing here knows where the prompt came from, so the same +// session is driven from the web page and from a chat platform. +// +// Driving an agent this way runs the real agent, with the configuration and the +// hooks Gateway already wrote into it. Its transcript, its usage and its tool +// permissions therefore keep working exactly as they do when somebody types at +// it, and none of that is reimplemented here. +package agentsession + +import ( + "strings" + "time" +) + +// State is where a session is in its turn. +type State string + +const ( + StateIdle State = "idle" + StateRunning State = "running" + StateFailed State = "failed" +) + +// Spec is what a session is opened with. +type Spec struct { + AgentId string `json:"agentId"` + AgentPath string `json:"agentPath"` + Owner string `json:"owner"` + // WorkDir is the directory the agent runs in, which is the project it acts + // on. Empty means the home of the account Gateway runs as. + WorkDir string `json:"workDir"` + // Model overrides the agent's own default for this session, empty to leave + // it alone. + Model string `json:"model"` + // Source names who opened the session - "web", or "im:<platform>:<user>" - + // and is what the records of this session are attributed to. + Source string `json:"source"` +} + +// Session is one conversation with one installed agent. +type Session struct { + Id string `json:"id"` + Spec + + // NativeId is the agent's own id for this conversation, which is what the + // next turn resumes. It is either handed to the agent when the session is + // opened, for one that takes it, or read out of the first turn's output. + NativeId string `json:"nativeId"` + // Resumable is false for an agent that carries nothing across turns, so the + // page can say that every message stands alone rather than pretending. + Resumable bool `json:"resumable"` + + Title string `json:"title"` + State State `json:"state"` + Turns int `json:"turns"` + CreatedTime string `json:"createdTime"` + UpdatedTime string `json:"updatedTime"` + // LastError is why the last turn failed, empty when it did not. + LastError string `json:"lastError,omitempty"` +} + +// titleLimit keeps a session's name to something a list and a chat window can +// both show on one line. +const titleLimit = 60 + +// titleFrom names a session after the first thing asked of it, which is what +// somebody looking at a list of them recognises. +func titleFrom(prompt string) string { + title := "" + for _, line := range strings.Split(prompt, "\n") { + if trimmed := strings.TrimSpace(line); trimmed != "" { + title = trimmed + break + } + } + if len([]rune(title)) > titleLimit { + title = string([]rune(title)[:titleLimit]) + "..." + } + return title +} + +func now() string { + return time.Now().Format(time.RFC3339Nano) +} diff --git a/agentsession/store.go b/agentsession/store.go new file mode 100644 index 0000000..7c4c3a8 --- /dev/null +++ b/agentsession/store.go @@ -0,0 +1,361 @@ +// Copyright 2026 The casbin Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package agentsession + +import ( + "context" + "errors" + "fmt" + "sync" + + "github.com/apache/casbin-gateway/agent" + "github.com/google/uuid" +) + +var ( + ErrNoSession = errors.New("no such session") + ErrNotDrivable = errors.New("this agent publishes no non-interactive mode, so Gateway cannot drive it") + ErrBusy = errors.New("this session is still answering") +) + +const ( + // historyLimit caps what one session replays to a page that connects while a + // turn is already running. A turn that says more than this is truncated at + // the front rather than kept whole in memory. + historyLimit = 4000 + // feedBuffer is how far one subscriber may fall behind before its events are + // dropped. Dropping is deliberate: a page nobody is looking at must not hold + // the agent up. + feedBuffer = 1024 +) + +// live is one session and whatever is listening to it. +type live struct { + mu sync.Mutex + session Session + events []Event + nextSeq int64 + subscribers map[int]chan Event + nextId int + cancel context.CancelFunc +} + +var registry = struct { + sync.Mutex + sessions map[string]*live +}{sessions: map[string]*live{}} + +// SessionSink is where sessions are kept, so they outlive the process. Storage +// lives outside this package for the same reason it does for the monitor: +// driving an agent must not depend on the database. +type SessionSink func(*Session) + +var sessionSink struct { + sync.RWMutex + sink SessionSink +} + +// SetSessionSink installs the store every change to a session is written to. +func SetSessionSink(sink SessionSink) { + sessionSink.Lock() + sessionSink.sink = sink + sessionSink.Unlock() +} + +func persist(session Session) { + sessionSink.RLock() + sink := sessionSink.sink + sessionSink.RUnlock() + if sink != nil { + sink(&session) + } +} + +// Open starts a session against one installation. Nothing runs yet: an agent is +// only started when there is something to ask it. +func Open(spec Spec) (Session, error) { + headless := agent.HeadlessOf(spec.AgentId) + if headless == nil { + return Session{}, ErrNotDrivable + } + + entry := &live{ + session: Session{ + Id: uuid.NewString(), + Spec: spec, + Resumable: headless.CanResume(), + State: StateIdle, + CreatedTime: now(), + UpdatedTime: now(), + }, + subscribers: map[int]chan Event{}, + } + // An agent that takes the id of the session it is starting is told ours, so + // the next turn resumes it without anything having to be read back. + if headless.NamesSession() { + entry.session.NativeId = entry.session.Id + } + + registry.Lock() + registry.sessions[entry.session.Id] = entry + registry.Unlock() + + persist(entry.session) + return entry.session, nil +} + +// Restore puts a session kept from an earlier run back in the registry, idle. +func Restore(session Session) { + session.State = StateIdle + entry := &live{session: session, subscribers: map[int]chan Event{}} + + registry.Lock() + registry.sessions[session.Id] = entry + registry.Unlock() +} + +func lookup(id string) (*live, error) { + registry.Lock() + entry, ok := registry.sessions[id] + registry.Unlock() + if !ok { + return nil, ErrNoSession + } + return entry, nil +} + +// Get is one session as it stands now. +func Get(id string) (Session, error) { + entry, err := lookup(id) + if err != nil { + return Session{}, err + } + + entry.mu.Lock() + defer entry.mu.Unlock() + return entry.session, nil +} + +// List is every session, newest first by the caller's own ordering. +func List() []Session { + registry.Lock() + entries := make([]*live, 0, len(registry.sessions)) + for _, entry := range registry.sessions { + entries = append(entries, entry) + } + registry.Unlock() + + result := make([]Session, 0, len(entries)) + for _, entry := range entries { + entry.mu.Lock() + result = append(result, entry.session) + entry.mu.Unlock() + } + return result +} + +// Close ends a session: whatever it is running is stopped, and it is dropped +// from the registry. What was stored of it is the caller's to remove. +func Close(id string) error { + entry, err := lookup(id) + if err != nil { + return err + } + + entry.mu.Lock() + if entry.cancel != nil { + entry.cancel() + } + for _, feed := range entry.subscribers { + close(feed) + } + entry.subscribers = map[int]chan Event{} + entry.mu.Unlock() + + registry.Lock() + delete(registry.sessions, id) + registry.Unlock() + return nil +} + +// Interrupt stops the turn a session is in the middle of, leaving the session +// itself open for the next one. +func Interrupt(id string) error { + entry, err := lookup(id) + if err != nil { + return err + } + + entry.mu.Lock() + defer entry.mu.Unlock() + if entry.cancel == nil { + return nil + } + entry.cancel() + return nil +} + +// Subscribe hands back what the session has said so far and a feed of what it +// says next, so a page that connects in the middle of a turn misses nothing. +// Only what came after "seen" is replayed, which is how a page that reconnects +// picks up where it left off instead of reading the conversation twice. +func Subscribe(id string, seen int64) (int, <-chan Event, []Event, error) { + entry, err := lookup(id) + if err != nil { + return 0, nil, nil, err + } + + entry.mu.Lock() + defer entry.mu.Unlock() + + entry.nextId++ + subscriberId := entry.nextId + feed := make(chan Event, feedBuffer) + entry.subscribers[subscriberId] = feed + + history := []Event{} + for _, event := range entry.events { + if event.Seq > seen { + history = append(history, event) + } + } + return subscriberId, feed, history, nil +} + +// Unsubscribe drops one feed. Closing is left to this side so that a publish +// racing with it cannot write to a closed channel. +func Unsubscribe(id string, subscriberId int) { + entry, err := lookup(id) + if err != nil { + return + } + + entry.mu.Lock() + defer entry.mu.Unlock() + if feed, ok := entry.subscribers[subscriberId]; ok { + delete(entry.subscribers, subscriberId) + close(feed) + } +} + +// publish records one event against the session and hands it to every listener. +func (entry *live) publish(event Event) { + entry.mu.Lock() + defer entry.mu.Unlock() + + entry.nextSeq++ + event.Seq = entry.nextSeq + entry.events = append(entry.events, event) + if len(entry.events) > historyLimit { + entry.events = entry.events[len(entry.events)-historyLimit:] + } + for _, feed := range entry.subscribers { + select { + case feed <- event: + default: + } + } +} + +// Send asks the session's agent one thing. It returns as soon as the agent has +// been started: the answer arrives on the feed. +func Send(id, prompt string) (Session, error) { + entry, err := lookup(id) + if err != nil { + return Session{}, err + } + + entry.mu.Lock() + if entry.session.State == StateRunning { + entry.mu.Unlock() + return Session{}, ErrBusy + } + + headless := agent.HeadlessOf(entry.session.AgentId) + if headless == nil { + entry.mu.Unlock() + return Session{}, ErrNotDrivable + } + + if entry.session.Title == "" { + entry.session.Title = titleFrom(prompt) + } + entry.session.State = StateRunning + entry.session.LastError = "" + entry.session.UpdatedTime = now() + session := entry.session + + ctx, cancel := context.WithCancel(context.Background()) + entry.cancel = cancel + entry.mu.Unlock() + + entry.publish(textEvent(EventPrompt, prompt)) + persist(session) + + go entry.turn(ctx, cancel, headless, prompt) + return session, nil +} + +// turn runs one exchange to its end and leaves the session ready for the next. +func (entry *live) turn(ctx context.Context, cancel context.CancelFunc, headless *agent.Headless, prompt string) { + defer cancel() + + entry.mu.Lock() + session := entry.session + entry.mu.Unlock() + + nativeId := "" + runErr := run(ctx, headless, session, prompt, func(event Event) { + if event.NativeId != "" { + nativeId = event.NativeId + } + entry.publish(event) + }) + + entry.mu.Lock() + entry.cancel = nil + entry.session.Turns++ + entry.session.UpdatedTime = now() + // An agent that names its own conversation only says so once it has started + // one, so the id is taken from the first turn that reported it. + if nativeId != "" && entry.session.NativeId == "" { + entry.session.NativeId = nativeId + } + if runErr != nil { + entry.session.State = StateFailed + entry.session.LastError = runErr.Error() + } else { + entry.session.State = StateIdle + } + session = entry.session + entry.mu.Unlock() + + if runErr != nil { + entry.publish(errorEvent(runErr)) + } + done := newEvent(EventDone) + done.NativeId = session.NativeId + entry.publish(done) + + persist(session) +} + +// Drivable reports whether an agent can be driven, and why not when it cannot. +func Drivable(agentId string) error { + if agent.HeadlessOf(agentId) == nil { + return fmt.Errorf("%s: %w", agent.DisplayNameOf(agentId), ErrNotDrivable) + } + return nil +} diff --git a/controllers/agent_session.go b/controllers/agent_session.go new file mode 100644 index 0000000..07455f9 --- /dev/null +++ b/controllers/agent_session.go @@ -0,0 +1,297 @@ +// Copyright 2026 The casbin Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package controllers + +import ( + "encoding/json" + "fmt" + "net/http" + "sort" + "strconv" + "strings" + "time" + + "github.com/apache/casbin-gateway/agent" + "github.com/apache/casbin-gateway/agentsession" + "github.com/apache/casbin-gateway/object" +) + +// sessionStreamHeartbeat keeps a stream that has gone quiet from being closed by +// whatever is between the page and Gateway. +const sessionStreamHeartbeat = 25 * time.Second + +// drivableAgent is one installation the chat page can open a session against. +type drivableAgent struct { + AgentId string `json:"agentId"` + Name string `json:"name"` + Path string `json:"path"` + Owner string `json:"owner"` + // Resumable is false for an agent that carries nothing from one turn to the + // next, which the page says out loud rather than letting somebody find out. + Resumable bool `json:"resumable"` +} + +// GetDrivableAgents lists the installations Gateway can hand a prompt to. An +// agent that publishes no non-interactive mode is left out rather than shown as +// something that will work later. +func (c *ApiController) GetDrivableAgents() { + if c.RequireAdmin() { + return + } + + installations, err := agent.Scan(false) + if err != nil { + c.ResponseError(err.Error()) + return + } + + result := []*drivableAgent{} + for _, installation := range installations { + headless := agent.HeadlessOf(installation.AgentId) + if headless == nil { + continue + } + result = append(result, &drivableAgent{ + AgentId: installation.AgentId, + Name: installation.Name, + Path: installation.Path, + Owner: installation.Owner, + Resumable: headless.CanResume(), + }) + } + c.ResponseOk(result) +} + +// GetDrivenSessions lists the conversations Gateway is driving, most recently +// used first. +func (c *ApiController) GetDrivenSessions() { + if c.RequireAdmin() { + return + } + + sessions := agentsession.List() + sort.Slice(sessions, func(i, j int) bool { + return sessions[i].UpdatedTime > sessions[j].UpdatedTime + }) + c.ResponseOk(sessions) +} + +// OpenDrivenSession starts a conversation against one installation. Nothing runs +// until it is asked something. +func (c *ApiController) OpenDrivenSession() { + if c.RequireAdmin() { + return + } + + var request struct { + AgentId string `json:"agentId"` + Path string `json:"path"` + Owner string `json:"owner"` + WorkDir string `json:"workDir"` + Model string `json:"model"` + } + if err := json.Unmarshal(c.Ctx.Input.RequestBody, &request); err != nil { + c.ResponseError(err.Error()) + return + } + + installation, err := findInstallation(request.AgentId, request.Path, request.Owner) + if err != nil { + c.ResponseError(err.Error()) + return + } + + session, err := agentsession.Open(agentsession.Spec{ + AgentId: installation.AgentId, + AgentPath: installation.Path, + Owner: installation.Owner, + WorkDir: strings.TrimSpace(request.WorkDir), + Model: strings.TrimSpace(request.Model), + Source: "web", + }) + if err != nil { + c.ResponseError(err.Error()) + return + } + c.ResponseOk(session) +} + +// SendDrivenSession asks the session's agent one thing and returns as soon as it +// has been started. What it says arrives on the stream. +func (c *ApiController) SendDrivenSession() { + if c.RequireAdmin() { + return + } + + var request struct { + Id string `json:"id"` + Prompt string `json:"prompt"` + } + if err := json.Unmarshal(c.Ctx.Input.RequestBody, &request); err != nil { + c.ResponseError(err.Error()) + return + } + if strings.TrimSpace(request.Prompt) == "" { + c.ResponseError("there is nothing to ask") + return + } + + session, err := agentsession.Send(request.Id, request.Prompt) + if err != nil { + c.ResponseError(err.Error()) + return + } + c.ResponseOk(session) +} + +// InterruptDrivenSession stops the turn a session is in the middle of, leaving +// the session open for the next one. +func (c *ApiController) InterruptDrivenSession() { + if c.RequireAdmin() { + return + } + + id := c.readSessionId() + if id == "" { + return + } + if err := agentsession.Interrupt(id); err != nil { + c.ResponseError(err.Error()) + return + } + c.ResponseOk() +} + +// CloseDrivenSession ends a conversation and forgets it. The agent keeps its own +// transcript of what was said, which is where the record of it lives. +func (c *ApiController) CloseDrivenSession() { + if c.RequireAdmin() { + return + } + + id := c.readSessionId() + if id == "" { + return + } + if err := agentsession.Close(id); err != nil { + c.ResponseError(err.Error()) + return + } + if err := object.DeleteAgentSession(id); err != nil { + c.ResponseError(err.Error()) + return + } + c.ResponseOk() +} + +// StreamDrivenSession pushes what a session says as it says it. What it has +// already said is replayed first, so a page that opens in the middle of a turn - +// or comes back to one - reads the whole conversation from here. +func (c *ApiController) StreamDrivenSession() { + if c.RequireAdmin() { + return + } + + id := c.Input().Get("id") + // The browser reconnects a dropped stream on its own and says which event it + // had last, so only what came after that is replayed. + seen, _ := strconv.ParseInt(firstNonEmpty(c.Ctx.Input.Header("Last-Event-ID"), c.Input().Get("seen")), 10, 64) + + subscriberId, feed, history, err := agentsession.Subscribe(id, seen) + if err != nil { + c.ResponseError(err.Error()) + return + } + defer agentsession.Unsubscribe(id, subscriberId) + + c.EnableRender = false + writer := c.Ctx.ResponseWriter + writer.Header().Set("Content-Type", "text/event-stream") + writer.Header().Set("Cache-Control", "no-cache") + writer.Header().Set("Connection", "keep-alive") + // An nginx in front of Gateway would otherwise hold every event back until + // the response ends. + writer.Header().Set("X-Accel-Buffering", "no") + writer.WriteHeader(http.StatusOK) + writer.Flush() + + for _, event := range history { + if !writeSessionEvent(writer, event) { + return + } + } + + ticker := time.NewTicker(sessionStreamHeartbeat) + defer ticker.Stop() + closed := c.Ctx.Request.Context().Done() + for { + select { + case event, ok := <-feed: + if !ok { + return + } + if !writeSessionEvent(writer, event) { + return + } + case <-ticker.C: + if _, err := fmt.Fprint(writer, ": keep-alive\n\n"); err != nil { + return + } + writer.Flush() + case <-closed: + return + } + } +} + +func writeSessionEvent(writer interface { + Write([]byte) (int, error) + Flush() +}, event agentsession.Event) bool { + data, err := json.Marshal(event) + if err != nil { + return true + } + if _, err := fmt.Fprintf(writer, "id: %d\nevent: message\ndata: %s\n\n", event.Seq, data); err != nil { + return false + } + writer.Flush() + return true +} + +func firstNonEmpty(values ...string) string { + for _, value := range values { + if value != "" { + return value + } + } + return "" +} + +// readSessionId takes the session out of a request that names nothing else. +func (c *ApiController) readSessionId() string { + var request struct { + Id string `json:"id"` + } + if err := json.Unmarshal(c.Ctx.Input.RequestBody, &request); err != nil { + c.ResponseError(err.Error()) + return "" + } + if request.Id == "" { + c.ResponseError("no session was named") + return "" + } + return request.Id +} diff --git a/main.go b/main.go index e1e0f41..c9460d9 100644 --- a/main.go +++ b/main.go @@ -125,6 +125,10 @@ func main() { } defer agentmonitor.Stop() + // The conversations Gateway drives on somebody's behalf, put back where a + // restart left them. + object.InitAgentSessions() + // Monitoring is on by default, so the agents already on this host are // patched without anyone opening the UI. The scan walks the disk, hence the // goroutine. diff --git a/object/agent_session.go b/object/agent_session.go new file mode 100644 index 0000000..9c7335f --- /dev/null +++ b/object/agent_session.go @@ -0,0 +1,149 @@ +// Copyright 2026 The casbin Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package object + +import ( + "sort" + + "github.com/apache/casbin-gateway/agentsession" + "github.com/beego/beego" + "github.com/xorm-io/core" +) + +// AgentSession is one conversation Gateway is driving, kept so that it outlives +// a restart: the agent's own id for it is what a later turn resumes, and losing +// that would strand the conversation inside the agent. +// +// What was said is not stored here. The agent keeps its own transcript, which +// the monitor already reads, and a second copy would only be one more place for +// somebody's prompts to sit. +type AgentSession struct { + Owner string `xorm:"varchar(100) notnull pk" json:"owner"` + Name string `xorm:"varchar(100) notnull pk" json:"name"` + + AgentId string `xorm:"varchar(100) index" json:"agentId"` + AgentPath string `xorm:"varchar(500)" json:"agentPath"` + AgentUser string `xorm:"varchar(100)" json:"agentUser"` + WorkDir string `xorm:"varchar(500)" json:"workDir"` + Model string `xorm:"varchar(200)" json:"model"` + // Source names who opened the session: "web", or "im:<platform>:<user>". + Source string `xorm:"varchar(200)" json:"source"` + + // NativeId is the agent's own id for the conversation, which is the whole + // reason a session is stored at all. + NativeId string `xorm:"varchar(200)" json:"nativeId"` + Resumable bool `json:"resumable"` + + Title string `xorm:"varchar(200)" json:"title"` + Turns int `json:"turns"` + LastError string `xorm:"varchar(1000)" json:"lastError"` + CreatedTime string `xorm:"varchar(100)" json:"createdTime"` + UpdatedTime string `xorm:"varchar(100) index" json:"updatedTime"` +} + +// agentSessionOwner is who every session belongs to. Gateway serves one machine, +// so that is its admin, the same as an agent's own row. +const agentSessionOwner = AgentOwner + +func toAgentSession(session *agentsession.Session) *AgentSession { + return &AgentSession{ + Owner: agentSessionOwner, + Name: session.Id, + AgentId: session.AgentId, + AgentPath: session.AgentPath, + AgentUser: session.Owner, + WorkDir: session.WorkDir, + Model: session.Model, + Source: session.Source, + NativeId: session.NativeId, + Resumable: session.Resumable, + Title: session.Title, + Turns: session.Turns, + LastError: session.LastError, + CreatedTime: session.CreatedTime, + UpdatedTime: session.UpdatedTime, + } +} + +func (session *AgentSession) toDriven() agentsession.Session { + return agentsession.Session{ + Id: session.Name, + Spec: agentsession.Spec{ + AgentId: session.AgentId, + AgentPath: session.AgentPath, + Owner: session.AgentUser, + WorkDir: session.WorkDir, + Model: session.Model, + Source: session.Source, + }, + NativeId: session.NativeId, + Resumable: session.Resumable, + Title: session.Title, + Turns: session.Turns, + LastError: session.LastError, + CreatedTime: session.CreatedTime, + UpdatedTime: session.UpdatedTime, + } +} + +// SaveAgentSession writes one session, inserting it the first time. It is the +// sink the driver hands every change to, so it must not fail loudly: a session +// that cannot be stored still answers, it just cannot be resumed after a +// restart. +func SaveAgentSession(session *agentsession.Session) { + row := toAgentSession(session) + affected, err := ormer.Engine.ID(core.PK{row.Owner, row.Name}).AllCols().Update(row) + if err == nil && affected == 0 { + _, err = ormer.Engine.Insert(row) + } + if err != nil { + beego.Error("agent session could not be stored:", err) + } +} + +// DeleteAgentSession removes one stored session. +func DeleteAgentSession(id string) error { + _, err := ormer.Engine.ID(core.PK{agentSessionOwner, id}).Delete(&AgentSession{}) + return err +} + +// GetAgentSessions is every stored session, most recently used first. +func GetAgentSessions() ([]*AgentSession, error) { + sessions := []*AgentSession{} + if err := ormer.Engine.Where("owner = ?", agentSessionOwner).Find(&sessions); err != nil { + return nil, err + } + + sort.Slice(sessions, func(i, j int) bool { + return sessions[i].UpdatedTime > sessions[j].UpdatedTime + }) + return sessions, nil +} + +// InitAgentSessions puts the stored sessions back in the driver and starts +// keeping it up to date. Nothing is running after this: a restored session only +// holds the agent's own id, and starts an agent again when it is next asked +// something. +func InitAgentSessions() { + sessions, err := GetAgentSessions() + if err != nil { + panic(err) + } + for _, session := range sessions { + agentsession.Restore(session.toDriven()) + } + + agentsession.SetSessionSink(SaveAgentSession) +} diff --git a/object/ormer.go b/object/ormer.go index 4514eb1..3e58bce 100644 --- a/object/ormer.go +++ b/object/ormer.go @@ -337,4 +337,11 @@ func (a *Ormer) createTable() { if err != nil { panic(err) } + + // The conversations Gateway is driving on somebody's behalf, kept for the + // agent's own id for each, which is what carries one across a restart. + err = a.Engine.Sync2(new(AgentSession)) + if err != nil { + panic(err) + } } diff --git a/routers/router.go b/routers/router.go index 6cf5a7f..b1fcef4 100644 --- a/routers/router.go +++ b/routers/router.go @@ -88,6 +88,15 @@ func initAPI() { beego.Router("/api/start-agent-instance", &controllers.ApiController{}, "POST:StartAgentInstance") beego.Router("/api/stop-agent-instance", &controllers.ApiController{}, "POST:StopAgentInstance") beego.Router("/api/capture-agent-instance-link", &controllers.ApiController{}, "POST:CaptureAgentInstanceLink") + + // Driving an agent: Gateway hands it a prompt and reads back what it says. + beego.Router("/api/get-drivable-agents", &controllers.ApiController{}, "GET:GetDrivableAgents") + beego.Router("/api/get-driven-sessions", &controllers.ApiController{}, "GET:GetDrivenSessions") + beego.Router("/api/open-driven-session", &controllers.ApiController{}, "POST:OpenDrivenSession") + beego.Router("/api/send-driven-session", &controllers.ApiController{}, "POST:SendDrivenSession") + beego.Router("/api/interrupt-driven-session", &controllers.ApiController{}, "POST:InterruptDrivenSession") + beego.Router("/api/close-driven-session", &controllers.ApiController{}, "POST:CloseDrivenSession") + beego.Router("/api/stream-driven-session", &controllers.ApiController{}, "GET:StreamDrivenSession") beego.Router("/api/get-agent-accounts", &controllers.ApiController{}, "GET:GetAgentAccounts") beego.Router("/api/save-agent-account", &controllers.ApiController{}, "POST:SaveAgentAccount") beego.Router("/api/add-agent-account", &controllers.ApiController{}, "POST:AddAgentAccount") diff --git a/web/src/App.tsx b/web/src/App.tsx index 16dc9e4..5c338ae 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -38,6 +38,7 @@ import AgentVersionsPage from "@/pages/AgentVersionsPage"; import AgentsPage from "@/pages/AgentsPage"; import AuthCallback from "@/pages/AuthCallback"; import AuthenticityPage from "@/pages/AuthenticityPage"; +import ChatPage from "@/pages/ChatPage"; import ConnectionsPage from "@/pages/ConnectionsPage"; import HomePage from "@/pages/HomePage"; import ImportLinkPage from "@/pages/ImportLinkPage"; @@ -286,6 +287,7 @@ export default function App() { )} /> <Route path="/account" element={requireSignin(user => <AccountPage account={user} />)} /> + <Route path="/chat" element={requireSignin(user => <ChatPage account={user} />)} /> <Route path="/agents" element={requireSignin(user => <AgentsPage account={user} />)} /> <Route path="/agent-versions" diff --git a/web/src/backend/DriveBackend.ts b/web/src/backend/DriveBackend.ts new file mode 100644 index 0000000..fc0eea4 --- /dev/null +++ b/web/src/backend/DriveBackend.ts @@ -0,0 +1,69 @@ +// Copyright 2026 The casbin Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +import {query, request} from "@/backend/request"; +import {ServerUrl} from "@/Setting"; +import type {DrivableAgent, DrivenEvent, DrivenSession} from "@/types"; + +/** The installations that publish a non-interactive mode, which is all Gateway can drive. */ +export function getDrivableAgents() { + return request<DrivableAgent[]>("/api/get-drivable-agents"); +} + +export function getDrivenSessions() { + return request<DrivenSession[]>("/api/get-driven-sessions"); +} + +/** Opens a conversation. Nothing runs until it is asked something. */ +export function openDrivenSession(target: {agentId: string; path: string; owner: string}, workDir: string, model = "") { + return request<DrivenSession>("/api/open-driven-session", "POST", {...target, workDir: workDir, model: model}); +} + +/** Asks one thing; the answer arrives on the stream, not in this response. */ +export function sendDrivenSession(id: string, prompt: string) { + return request<DrivenSession>("/api/send-driven-session", "POST", {id: id, prompt: prompt}); +} + +export function interruptDrivenSession(id: string) { + return request<boolean>("/api/interrupt-driven-session", "POST", {id: id}); +} + +export function closeDrivenSession(id: string) { + return request<boolean>("/api/close-driven-session", "POST", {id: id}); +} + +/** + * Opens the feed of one conversation; the browser reconnects on its own and the + * server replays only what was missed. Returns the function that closes it. + */ +export function streamDrivenSession( + id: string, + handlers: {onEvent: (event: DrivenEvent) => void; onOpen?: () => void; onError?: () => void}, +) { + const source = new EventSource(`${ServerUrl}/api/stream-driven-session${query({id: id})}`, { + withCredentials: true, + }); + + source.addEventListener("message", event => { + try { + handlers.onEvent(JSON.parse((event as MessageEvent).data) as DrivenEvent); + } catch { + // A malformed event is not worth tearing the feed down for. + } + }); + source.onopen = () => handlers.onOpen?.(); + source.onerror = () => handlers.onError?.(); + + return () => source.close(); +} diff --git a/web/src/locales/en/data.json b/web/src/locales/en/data.json index 487c024..cd33393 100644 --- a/web/src/locales/en/data.json +++ b/web/src/locales/en/data.json @@ -831,6 +831,21 @@ "What was sent": "What was sent", "Wrong": "Wrong" }, + "chat": { + "Ask an agent on this machine something, and watch it work": "Ask an agent on this machine something, and watch it work", + "Ask the agent something": "Ask the agent something", + "Chat": "Chat", + "Each message stands alone": "Each message stands alone", + "New session": "New session", + "No agent on this machine can be driven": "No agent on this machine can be driven", + "No session is open": "No session is open", + "Nothing asked yet": "Nothing asked yet", + "Pick an agent": "Pick an agent", + "Thinking": "Thinking", + "Working directory": "Working directory", + "{count} tokens in": "{count} tokens in", + "{count} tokens out": "{count} tokens out" + }, "connector": { "API key": "API key", "All": "All", diff --git a/web/src/locales/zh/data.json b/web/src/locales/zh/data.json index 8f92e19..6654552 100644 --- a/web/src/locales/zh/data.json +++ b/web/src/locales/zh/data.json @@ -903,6 +903,21 @@ "case.vendor-headers.method": "这一条只向厂商自己的域名发问，比对的也是该厂商文档里写明的响应头：能说OpenAI兼容的API，不等于自称是OpenAI。真货前面的转售商通常会透传一部分，而从没连过厂商的后端一个也拿不出来；但把响应头抹掉也可能只是干净，所以这一条单独不会判定为异常。响应头列表留空就用该厂商自己的，也可以在这里指定要检查哪些响应头。", "case.vendor-headers.question": "厂商自己API的那些响应头回来了吗？" }, + "chat": { + "Ask an agent on this machine something, and watch it work": "让这台机器上的Agent做事，并看着它做", + "Ask the agent something": "问这个Agent点什么", + "Chat": "聊天", + "Each message stands alone": "每条消息各自独立", + "New session": "新建会话", + "No agent on this machine can be driven": "这台机器上没有可驱动的Agent", + "No session is open": "没有打开的会话", + "Nothing asked yet": "还没问过什么", + "Pick an agent": "选一个Agent", + "Thinking": "思考过程", + "Working directory": "工作目录", + "{count} tokens in": "输入{count}个token", + "{count} tokens out": "输出{count}个token" + }, "connector": { "A browser that runs in Browserbase's cloud rather than on this machine, for pages that need a real one and a clean address.": "跑在Browserbase云端而不是本机的浏览器，适合那些需要真实浏览器和干净出口IP的页面。", "A knowledge graph the agent writes to and reads back, so what it learned in one session is there in the next.": "一张Agent可以写入和回读的知识图谱，这次会话学到的东西下次还在。", diff --git a/web/src/nav.ts b/web/src/nav.ts index 2bc0b2f..396797f 100644 --- a/web/src/nav.ts +++ b/web/src/nav.ts @@ -25,6 +25,7 @@ import { PackageCheck, Plug, Route, + SendHorizontal, Settings, ShieldCheck, ShieldHalf, @@ -69,6 +70,9 @@ export const navGroups: NavGroup[] = [ // The table of installations is the home screen's own advanced view, linked // from it rather than competing with it in the rail. {key: "/agents", label: "agent:Advanced view", icon: Table2, path: "/agents", adminOnly: true, hidden: true}, + // Driving an agent is what somebody does with one, so it sits beside the + // list of them rather than down among the pages that only report. + {key: "/chat", label: "chat:Chat", icon: SendHorizontal, path: "/chat", adminOnly: true}, { key: "/agent-versions", label: "agent:Agent versions", diff --git a/web/src/pages/ChatPage.tsx b/web/src/pages/ChatPage.tsx new file mode 100644 index 0000000..acd5fb8 --- /dev/null +++ b/web/src/pages/ChatPage.tsx @@ -0,0 +1,441 @@ +// Copyright 2026 The casbin Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +import * as React from "react"; +import {ArrowUp, MessageSquare, Plus, Square, Trash2, Wrench} from "lucide-react"; +import i18next from "i18next"; + +import * as DriveBackend from "@/backend/DriveBackend"; +import * as Setting from "@/Setting"; +import {AgentIcon} from "@/components/AgentIcon"; +import {EmptyState} from "@/components/shared/empty-state"; +import {Fold} from "@/components/shared/fold"; +import {Markdown} from "@/components/shared/markdown"; +import {PageContainer, PageHeader} from "@/components/shared/page-header"; +import {SimpleSelect} from "@/components/shared/simple-select"; +import {UnauthorizedResult} from "@/components/shared/misc"; +import {MessageAlert} from "@/components/ui/alert"; +import {Badge} from "@/components/ui/badge"; +import {Button} from "@/components/ui/button"; +import {Input} from "@/components/ui/input"; +import {Textarea} from "@/components/ui/textarea"; +import {cn} from "@/lib/utils"; +import type {Account, DrivableAgent, DrivenEvent, DrivenSession} from "@/types"; + +/** One tool the agent reached for, and what came back. */ +interface ToolBlock { + kind: "tool"; + seq: number; + name: string; + input: string; + result?: string; +} + +type Block = {kind: "text" | "thinking"; seq: number; text: string} | ToolBlock; + +/** One exchange: what was asked, and everything said before the next question. */ +interface Turn { + seq: number; + prompt: string; + blocks: Block[]; + error: string; + usage?: DrivenEvent; + done: boolean; +} + +/** + * Flattens the event feed into exchanges. Consecutive pieces of prose are joined + * so a streamed answer reads as one paragraph rather than as the fragments the + * agent happened to send it in. + */ +function toTurns(events: DrivenEvent[]): Turn[] { + const turns: Turn[] = []; + let turn: Turn | null = null; + + const append = (kind: "text" | "thinking", event: DrivenEvent) => { + if (!turn || !event.text) { + return; + } + const last = turn.blocks[turn.blocks.length - 1]; + if (last && last.kind === kind) { + last.text += event.text; + return; + } + turn.blocks.push({kind: kind, seq: event.seq, text: event.text}); + }; + + for (const event of events) { + switch (event.type) { + case "prompt": + turn = {seq: event.seq, prompt: event.text ?? "", blocks: [], error: "", done: false}; + turns.push(turn); + break; + case "text": + append("text", event); + break; + case "thinking": + append("thinking", event); + break; + case "toolUse": + turn?.blocks.push({ + kind: "tool", + seq: event.seq, + name: event.toolName ?? "", + input: event.toolInput ?? "", + }); + break; + case "toolResult": { + const tool = [...(turn?.blocks ?? [])].reverse().find(block => block.kind === "tool") as ToolBlock | undefined; + if (tool) { + tool.result = (tool.result ?? "") + (event.text ?? ""); + } + break; + } + case "usage": + if (turn && (event.inputTokens || event.outputTokens || event.costUsd)) { + turn.usage = event; + } + break; + case "error": + if (turn) { + turn.error = event.text ?? ""; + } + break; + case "done": + if (turn) { + turn.done = true; + } + break; + } + } + return turns; +} + +function ToolRow({block}: {block: ToolBlock}) { + return ( + <div className="bg-muted/40 rounded-md border text-xs"> + <div className="flex items-center gap-2 border-b px-2.5 py-1.5"> + <Wrench className="text-muted-foreground size-3.5 shrink-0" /> + <span className="font-medium">{block.name}</span> + </div> + <pre className="max-h-40 overflow-auto px-2.5 py-1.5 font-mono text-[11px] whitespace-pre-wrap">{block.input}</pre> + {block.result ? ( + <pre className="text-muted-foreground max-h-40 overflow-auto border-t px-2.5 py-1.5 font-mono text-[11px] whitespace-pre-wrap"> + {block.result} + </pre> + ) : null} + </div> + ); +} + +function TurnView({turn}: {turn: Turn}) { + return ( + <div className="flex flex-col gap-3"> + <div className="flex justify-end"> + <div className="bg-primary/10 border-primary/20 max-w-[85%] rounded-lg border px-3 py-2 text-sm whitespace-pre-wrap"> + {turn.prompt} + </div> + </div> + + {turn.blocks.map(block => { + if (block.kind === "tool") { + return <ToolRow key={block.seq} block={block} />; + } + if (block.kind === "thinking") { + return ( + <Fold key={block.seq} title={i18next.t("chat:Thinking")}> + <div className="text-muted-foreground text-sm whitespace-pre-wrap italic">{block.text}</div> + </Fold> + ); + } + return ( + <div key={block.seq} className="text-sm"> + <Markdown content={block.text} /> + </div> + ); + })} + + {turn.error ? <MessageAlert variant="destructive" title={turn.error} /> : null} + + {turn.usage ? ( + <div className="text-muted-foreground flex flex-wrap items-center gap-2 text-xs"> + <Badge variant="outline"> + {i18next.t("chat:{count} tokens in").replace("{count}", String(turn.usage.inputTokens ?? 0))} + </Badge> + <Badge variant="outline"> + {i18next.t("chat:{count} tokens out").replace("{count}", String(turn.usage.outputTokens ?? 0))} + </Badge> + {turn.usage.costUsd ? <Badge variant="outline">${turn.usage.costUsd.toFixed(4)}</Badge> : null} + </div> + ) : null} + </div> + ); +} + +/** + * Driving an agent from Gateway. The agent that runs is the real one, with the + * configuration and the hooks already written into it, so what it costs and what + * it was allowed to do are recorded exactly as when somebody types at it. + */ +export default function ChatPage({account}: {account: Account}) { + const isAdmin = Setting.isAdminUser(account); + const [agents, setAgents] = React.useState<DrivableAgent[]>([]); + const [sessions, setSessions] = React.useState<DrivenSession[]>([]); + const [activeId, setActiveId] = React.useState(""); + const [events, setEvents] = React.useState<DrivenEvent[]>([]); + const [draft, setDraft] = React.useState(""); + const [target, setTarget] = React.useState(""); + const [workDir, setWorkDir] = React.useState(""); + const [error, setError] = React.useState(""); + const [loading, setLoading] = React.useState(true); + const bottom = React.useRef<HTMLDivElement>(null); + + const active = sessions.find(session => session.id === activeId); + const running = active?.state === "running"; + + const loadSessions = React.useCallback(() => { + return DriveBackend.getDrivenSessions() + .then(res => { + if (res.status === "ok") { + setSessions(res.data ?? []); + } else { + setError(res.msg || i18next.t("general:Failed to get data")); + } + }) + .catch(failure => setError(failure.message || String(failure))); + }, []); + + React.useEffect(() => { + if (!isAdmin) { + setLoading(false); + return; + } + Promise.all([ + DriveBackend.getDrivableAgents().then(res => { + if (res.status === "ok") { + const found = res.data ?? []; + setAgents(found); + setTarget(previous => previous || (found[0] ? found[0].agentId + "\n" + found[0].path : "")); + } + }), + loadSessions(), + ]).then(() => setLoading(false)); + }, [isAdmin, loadSessions]); + + // One stream per session: the server replays what was already said, so the + // conversation is read from the feed rather than fetched separately. + React.useEffect(() => { + if (!activeId) { + setEvents([]); + return; + } + setEvents([]); + const close = DriveBackend.streamDrivenSession(activeId, { + onEvent: event => { + setEvents(previous => (previous.some(seen => seen.seq === event.seq) ? previous : [...previous, event])); + if (event.type === "done") { + loadSessions(); + } + }, + }); + return close; + }, [activeId, loadSessions]); + + React.useEffect(() => { + bottom.current?.scrollIntoView({behavior: "smooth"}); + }, [events.length]); + + const open = () => { + const [agentId, path] = target.split("\n"); + const installation = agents.find(candidate => candidate.agentId === agentId && candidate.path === path); + if (!installation) { + return; + } + DriveBackend.openDrivenSession( + {agentId: installation.agentId, path: installation.path, owner: installation.owner}, + workDir.trim(), + ) + .then(res => { + if (res.status === "ok" && res.data) { + setSessions(previous => [res.data as DrivenSession, ...previous]); + setActiveId((res.data as DrivenSession).id); + setError(""); + } else { + setError(res.msg || i18next.t("general:Failed to save")); + } + }) + .catch(failure => setError(failure.message || String(failure))); + }; + + const send = () => { + const prompt = draft.trim(); + if (!prompt || !activeId || running) { + return; + } + setDraft(""); + DriveBackend.sendDrivenSession(activeId, prompt) + .then(res => { + if (res.status === "ok") { + loadSessions(); + } else { + setError(res.msg || i18next.t("general:Failed to save")); + } + }) + .catch(failure => setError(failure.message || String(failure))); + }; + + const close = (id: string) => { + DriveBackend.closeDrivenSession(id).then(() => { + setSessions(previous => previous.filter(session => session.id !== id)); + setActiveId(previous => (previous === id ? "" : previous)); + }); + }; + + if (!isAdmin) { + return <UnauthorizedResult />; + } + + const turns = toTurns(events); + + return ( + <PageContainer> + <PageHeader + title={i18next.t("chat:Chat")} + description={i18next.t("chat:Ask an agent on this machine something, and watch it work")} + /> + + {error ? <MessageAlert variant="destructive" title={error} /> : null} + + <div className="grid gap-4 lg:grid-cols-[300px_1fr]"> + <div className="flex flex-col gap-3"> + <div className="flex flex-col gap-2 rounded-lg border p-3"> + <SimpleSelect + value={target} + onChange={setTarget} + size="sm" + placeholder={i18next.t("chat:Pick an agent")} + options={agents.map(candidate => ({ + value: candidate.agentId + "\n" + candidate.path, + text: candidate.name, + label: ( + <span className="flex items-center gap-2"> + <AgentIcon agent={candidate.name} size={16} /> + <span className="truncate">{candidate.name}</span> + </span> + ), + }))} + /> + <Input + value={workDir} + onChange={event => setWorkDir(event.target.value)} + placeholder={i18next.t("chat:Working directory")} + className="h-8 text-xs" + /> + <Button size="sm" onClick={open} disabled={!target}> + <Plus className="size-4" /> + {i18next.t("chat:New session")} + </Button> + </div> + + <div className="flex flex-col gap-1"> + {sessions.map(session => ( + <button + key={session.id} + type="button" + onClick={() => setActiveId(session.id)} + className={cn( + "group hover:bg-muted/60 flex items-center gap-2 rounded-md border border-transparent px-2 py-1.5 text-left text-sm", + session.id === activeId && "bg-muted border-border", + )} + > + <AgentIcon agent={session.agentId} size={16} /> + <span className="min-w-0 flex-1 truncate"> + {session.title || i18next.t("chat:New session")} + {session.state === "running" ? <span className="text-muted-foreground"> ...</span> : null} + </span> + <Trash2 + className="text-muted-foreground hover:text-destructive size-3.5 shrink-0 opacity-0 group-hover:opacity-100" + onClick={event => { + event.stopPropagation(); + close(session.id); + }} + /> + </button> + ))} + </div> + </div> + + <div className="flex min-h-[60vh] min-w-0 flex-col rounded-lg border"> + {active ? ( + <> + <div className="text-muted-foreground flex items-center gap-2 border-b px-3 py-2 text-xs"> + <AgentIcon agent={active.agentId} size={14} /> + <span className="shrink-0 font-medium">{active.agentId}</span> + <span className="min-w-0 flex-1 truncate font-mono" title={active.workDir}> + {active.workDir || "~"} + </span> + {active.resumable ? null : <Badge variant="outline">{i18next.t("chat:Each message stands alone")}</Badge>} + </div> + + <div className="flex-1 overflow-y-auto px-3 py-4"> + {turns.length === 0 ? ( + <EmptyState icon={MessageSquare} title={i18next.t("chat:Nothing asked yet")} /> + ) : ( + <div className="flex flex-col gap-6"> + {turns.map(turn => ( + <TurnView key={turn.seq} turn={turn} /> + ))} + </div> + )} + <div ref={bottom} /> + </div> + + <div className="flex items-end gap-2 border-t p-2"> + <Textarea + value={draft} + onChange={event => setDraft(event.target.value)} + onKeyDown={event => { + if (event.key === "Enter" && !event.shiftKey) { + event.preventDefault(); + send(); + } + }} + placeholder={i18next.t("chat:Ask the agent something")} + className="max-h-40 min-h-10 resize-none" + /> + {running ? ( + <Button size="icon" variant="outline" onClick={() => DriveBackend.interruptDrivenSession(activeId)}> + <Square className="size-4" /> + </Button> + ) : ( + <Button size="icon" onClick={send} disabled={!draft.trim()}> + <ArrowUp className="size-4" /> + </Button> + )} + </div> + </> + ) : ( + <EmptyState + icon={MessageSquare} + title={loading ? i18next.t("general:Loading") : i18next.t("chat:No session is open")} + description={ + agents.length === 0 && !loading ? i18next.t("chat:No agent on this machine can be driven") : undefined + } + /> + )} + </div> + </div> + </PageContainer> + ); +} diff --git a/web/src/types.ts b/web/src/types.ts index 1fde7d3..b5d138b 100644 --- a/web/src/types.ts +++ b/web/src/types.ts @@ -522,6 +522,52 @@ export interface AgentSession { historical?: boolean; } +/** One installation Gateway can hand a prompt to. */ +export interface DrivableAgent { + agentId: string; + name: string; + path: string; + owner: string; + /** False for an agent that carries nothing from one turn to the next. */ + resumable: boolean; +} + +/** One conversation Gateway is driving on somebody's behalf. */ +export interface DrivenSession { + id: string; + agentId: string; + agentPath: string; + owner: string; + workDir: string; + model: string; + source: string; + nativeId: string; + resumable: boolean; + title: string; + state: "idle" | "running" | "failed"; + turns: number; + createdTime: string; + updatedTime: string; + lastError?: string; +} + +/** One piece of what a driven agent said, in the shape every agent is flattened into. */ +export interface DrivenEvent { + type: "prompt" | "text" | "thinking" | "toolUse" | "toolResult" | "usage" | "done" | "error"; + seq: number; + createdTime: string; + text?: string; + toolName?: string; + toolUseId?: string; + toolInput?: string; + model?: string; + nativeId?: string; + inputTokens?: number; + outputTokens?: number; + costUsd?: number; + durationMs?: number; +} + /** One part of a transcript message: prose, thinking, a tool call or its result. */ export interface AgentMessageBlock { kind: "text" | "thinking" | "toolUse" | "toolResult" | "image"; --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
