diff --git a/forge-cli/cmd/serve.go b/forge-cli/cmd/serve.go index 84c99009..9481150a 100644 --- a/forge-cli/cmd/serve.go +++ b/forge-cli/cmd/serve.go @@ -91,21 +91,24 @@ var serveStartCmd = &cobra.Command{ } var serveStopCmd = &cobra.Command{ - Use: "stop", - Short: "Stop the running agent daemon", - RunE: serveStopRun, + Use: "stop", + Short: "Stop the running agent daemon", + SilenceUsage: true, // a failed stop shouldn't bury the error in flag help + RunE: serveStopRun, } var serveStatusCmd = &cobra.Command{ - Use: "status", - Short: "Show agent daemon status", - RunE: serveStatusRun, + Use: "status", + Short: "Show agent daemon status", + SilenceUsage: true, + RunE: serveStatusRun, } var serveLogsCmd = &cobra.Command{ - Use: "logs", - Short: "Tail daemon log output", - RunE: serveLogsRun, + Use: "logs", + Short: "Tail daemon log output", + SilenceUsage: true, + RunE: serveLogsRun, } func registerServeFlags(cmd *cobra.Command) { @@ -370,6 +373,16 @@ func serveStopRun(cmd *cobra.Command, args []string) error { fmt.Fprintln(os.Stderr, "Daemon did not stop in time, sending SIGKILL...") if err := sendKillSignal(proc); err != nil { os.Remove(statePath) //nolint:errcheck + + // The daemon can exit between the last liveness check and this + // kill, and killing an already-dead process is an error on both + // platforms (Windows reports "Access is denied"). The goal was + // for it to be gone, and it is — that's a successful stop. + if !process.IsAlive(state.PID) { + fmt.Fprintln(os.Stderr, "Daemon stopped.") + return nil + } + return fmt.Errorf("sending SIGKILL: %w", err) } diff --git a/forge-core/util/process/process_windows.go b/forge-core/util/process/process_windows.go index 91c575fd..5d7c0ef2 100644 --- a/forge-core/util/process/process_windows.go +++ b/forge-core/util/process/process_windows.go @@ -13,6 +13,14 @@ import "syscall" // rights than this check needs. const processQueryLimitedInfo = 0x1000 +// processSynchronize (SYNCHRONIZE) is required to wait on a process +// handle, which is how liveness is actually determined below. +const processSynchronize = 0x00100000 + +// waitTimeout (WAIT_TIMEOUT) is WaitForSingleObject's answer when the +// handle is NOT signaled — i.e. the process has not exited. +const waitTimeout = 0x00000102 + // IsAlive reports whether a process with the given PID is currently running. // // On Windows, the Unix idiom os.Process.Signal(syscall.Signal(0)) does not @@ -21,19 +29,34 @@ const processQueryLimitedInfo = 0x1000 // "operating system does not support signal". This always-error response // makes Signal(0) useless as a liveness probe on Windows. // -// Instead, open the process handle with PROCESS_QUERY_LIMITED_INFORMATION -// rights. OpenProcess fails only when the PID doesn't exist or the caller -// lacks rights even for the limited-info subset; both cases reasonably map -// to "not alive" for our use case (the forge daemon is the caller's child, -// so it always has rights to its own PID). +// Handle openability is NOT liveness either: Windows keeps a terminated +// process's kernel object alive as long as any handle to it remains open +// (so callers can still read its exit code), and OpenProcess succeeds +// against that object. A caller that holds a handle while polling — as +// `forge serve stop` does via os.FindProcess — would therefore see its +// own dead child reported as running forever, wait out the full timeout, +// and then fail with "Access is denied" trying to kill it twice. // -// The handle is closed immediately — we only care that OpenProcess -// succeeded, not what the handle exposes. +// So open the handle and wait on it with a zero timeout instead. A +// process handle becomes signaled exactly when the process exits, which +// is unambiguous: signaled means exited, WAIT_TIMEOUT means still +// running. (GetExitCodeProcess is the other option, but its STILL_ACTIVE +// sentinel is 259 and collides with a genuine exit code of 259.) func IsAlive(pid int) bool { - h, err := syscall.OpenProcess(processQueryLimitedInfo, false, uint32(pid)) + h, err := syscall.OpenProcess(processQueryLimitedInfo|processSynchronize, false, uint32(pid)) + if err != nil { + // PID doesn't exist, or we lack even limited-info rights — + // both map to "not alive" for our use case (the forge daemon + // is the caller's child, so rights are never the issue). + return false + } + defer syscall.CloseHandle(h) //nolint:errcheck + + event, err := syscall.WaitForSingleObject(h, 0) if err != nil { + // WAIT_FAILED. Report not-alive rather than pinning a caller + // in a poll loop it can never exit. return false } - _ = syscall.CloseHandle(h) - return true + return event == waitTimeout } diff --git a/forge-core/util/process/process_windows_test.go b/forge-core/util/process/process_windows_test.go new file mode 100644 index 00000000..03e084af --- /dev/null +++ b/forge-core/util/process/process_windows_test.go @@ -0,0 +1,57 @@ +//go:build windows + +package process + +import ( + "os/exec" + "syscall" + "testing" +) + +// Windows keeps a terminated process's kernel object alive while any +// handle to it stays open, so OpenProcess still succeeds against a dead +// PID. `forge serve stop` holds exactly such a handle (from +// os.FindProcess) while it polls IsAlive, which made an +// openability-based check report its own dead child as running until the +// timeout expired — then fail with "Access is denied" on the second kill. +// +// TestIsAlive_ChildAfterExit does not cover this: cmd.Wait() closes Go's +// handle, releasing the object before the check runs. The handle must be +// held across the kill to reproduce it. +func TestIsAlive_FalseWhileHandleHeld(t *testing.T) { + // ping runs long enough to be observed alive and needs no stdin + // (timeout/pause both fail when stdin isn't a console). + cmd := exec.Command("ping", "-n", "30", "127.0.0.1") + if err := cmd.Start(); err != nil { + t.Fatalf("starting subprocess: %v", err) + } + pid := cmd.Process.Pid + defer func() { + _ = cmd.Process.Kill() + _, _ = cmd.Process.Wait() + }() + + // Mirror os.FindProcess: an independent handle held across the kill. + h, err := syscall.OpenProcess(processQueryLimitedInfo|processSynchronize, false, uint32(pid)) + if err != nil { + t.Fatalf("OpenProcess: %v", err) + } + defer syscall.CloseHandle(h) //nolint:errcheck + + if !IsAlive(pid) { + t.Fatalf("IsAlive(pid=%d) = false while running, want true", pid) + } + + if err := cmd.Process.Kill(); err != nil { + t.Fatalf("Kill: %v", err) + } + if _, err := cmd.Process.Wait(); err != nil { + t.Fatalf("Wait: %v", err) + } + + // h is still open, so the PID remains openable. IsAlive must not be + // fooled by that. + if IsAlive(pid) { + t.Errorf("IsAlive(pid=%d) = true for a terminated process while a handle is held, want false", pid) + } +} diff --git a/forge-ui/handlers.go b/forge-ui/handlers.go index 64ed7f15..3cb4f3de 100644 --- a/forge-ui/handlers.go +++ b/forge-ui/handlers.go @@ -126,7 +126,7 @@ func (s *UIServer) handleStopAgent(w http.ResponseWriter, r *http.Request) { return } - if err := s.pm.Stop(id, agent.Directory); err != nil { + if err := s.pm.Stop(id, agent); err != nil { writeError(w, http.StatusConflict, err.Error()) return } diff --git a/forge-ui/process.go b/forge-ui/process.go index d828d9e8..933d599b 100644 --- a/forge-ui/process.go +++ b/forge-ui/process.go @@ -89,6 +89,23 @@ func NewProcessManager(exePath string, broker *SSEBroker, basePort int) *Process } } +// broadcastStatus emits a snapshot of info rather than the pointer itself. +// SSEBroker queues events and handleSSE marshals them when it drains the +// channel, so sharing the pointer would let a later mutation rewrite an +// already-queued event — a "stopping" event would serialize as "stopped". +// +// The snapshot is a SHALLOW copy: it decouples the scalar fields this package +// mutates (Status, Port, Error), but Tools, Channels and DeniedChannels still +// alias the caller's backing arrays, and StartedAt stays a shared pointer. +// That is safe only because Start/Stop never write through them — they are +// populated once by Scanner.scanDir and read-only thereafter. Mutating a slice +// element in place (rather than replacing the slice) would reintroduce the +// aliasing bug for that field, so deep-copy here if that ever changes. +func (pm *ProcessManager) broadcastStatus(info *AgentInfo) { + snapshot := *info + pm.broker.Broadcast(SSEEvent{Type: "agent_status", Data: &snapshot}) +} + // Start launches an agent via `forge serve start`. func (pm *ProcessManager) Start(agentID string, info *AgentInfo, passphrase string) error { pm.mu.Lock() @@ -124,7 +141,7 @@ func (pm *ProcessManager) Start(agentID string, info *AgentInfo, passphrase stri info.Status = StateErrored info.Error = errMsg - pm.broker.Broadcast(SSEEvent{Type: "agent_status", Data: info}) + pm.broadcastStatus(info) return fmt.Errorf("forge serve start failed: %s", errMsg) } @@ -144,7 +161,7 @@ func (pm *ProcessManager) Start(agentID string, info *AgentInfo, passphrase stri info.Status = StateErrored info.Error = errMsg - pm.broker.Broadcast(SSEEvent{Type: "agent_status", Data: info}) + pm.broadcastStatus(info) return fmt.Errorf("agent failed to start: %s", errMsg) } @@ -152,7 +169,7 @@ func (pm *ProcessManager) Start(agentID string, info *AgentInfo, passphrase stri info.Status = StateRunning info.Port = port info.Error = "" - pm.broker.Broadcast(SSEEvent{Type: "agent_status", Data: info}) + pm.broadcastStatus(info) return nil } @@ -238,15 +255,41 @@ func (pm *ProcessManager) readServeLogs(agentDir string) string { } // Stop stops an agent via `forge serve stop`. -func (pm *ProcessManager) Stop(agentID string, agentDir string) error { +// +// Takes the full *AgentInfo so every broadcast carries a complete record: +// fields without omitempty marshal as zero values, and the dashboard merges +// events by object spread, so a stub event blanks the card's metadata. +func (pm *ProcessManager) Stop(agentID string, info *AgentInfo) error { pm.mu.Lock() defer pm.mu.Unlock() + // Announce before the blocking cmd.Run() — the HTTP handler won't + // respond until it returns, so this is the only thing that can flip + // the card to "stopping". SSE is served on its own goroutine. + prevStatus, prevPort := info.Status, info.Port + info.Status = StateStopping + pm.broadcastStatus(info) + cmd := exec.Command(pm.exePath, "serve", "stop") - cmd.Dir = agentDir + cmd.Dir = info.Directory + + var stderr bytes.Buffer + cmd.Stderr = &stderr if err := cmd.Run(); err != nil { - return fmt.Errorf("forge serve stop failed: %w", err) + errMsg := strings.TrimSpace(stderr.String()) + if errMsg == "" { + errMsg = err.Error() + } + + // Roll back, or the card stays wedged on "stopping" with its + // buttons disabled. + info.Status = prevStatus + info.Port = prevPort + info.Error = errMsg + pm.broadcastStatus(info) + + return fmt.Errorf("forge serve stop failed: %s", errMsg) } if port, ok := pm.allocated[agentID]; ok { @@ -254,11 +297,11 @@ func (pm *ProcessManager) Stop(agentID string, agentDir string) error { delete(pm.allocated, agentID) } - pm.broker.Broadcast(SSEEvent{Type: "agent_status", Data: &AgentInfo{ - ID: agentID, - Directory: agentDir, - Status: StateStopped, - }}) + // Clear the port — otherwise a stopped card still shows a port tag. + info.Status = StateStopped + info.Port = 0 + info.Error = "" + pm.broadcastStatus(info) return nil } diff --git a/forge-ui/process_test.go b/forge-ui/process_test.go index bc2d23c5..103fa244 100644 --- a/forge-ui/process_test.go +++ b/forge-ui/process_test.go @@ -57,10 +57,68 @@ func TestProcessManagerStopExecError(t *testing.T) { broker := NewSSEBroker() pm := NewProcessManager("/nonexistent/binary", broker, 9100) - err := pm.Stop("nonexistent", t.TempDir()) + info := &AgentInfo{ + ID: "nonexistent", + Directory: t.TempDir(), + Status: StateRunning, + Port: 9100, + } + + err := pm.Stop("nonexistent", info) if err == nil { t.Error("expected error stopping with non-existent binary") } + + // A failed stop must roll back, not leave the card on "stopping". + if info.Status != StateRunning { + t.Errorf("status = %q, want %q after failed stop", info.Status, StateRunning) + } + if info.Port != 9100 { + t.Errorf("port = %d, want 9100 preserved after failed stop", info.Port) + } + if info.Error == "" { + t.Error("expected Error to be populated after failed stop") + } +} + +// Stop must emit a "stopping" event BEFORE it blocks on `forge serve stop`, +// and every event must carry a full record — a stub would blank the card's +// metadata in the dashboard. +func TestProcessManagerStopBroadcastsStoppingFirst(t *testing.T) { + broker := NewSSEBroker() + ch := broker.Subscribe() + defer broker.Unsubscribe(ch) + + pm := NewProcessManager("/nonexistent/binary", broker, 9100) + + info := &AgentInfo{ + ID: "test-agent", + Version: "1.2.3", + Framework: "langchain", + Channels: []string{"slack"}, + Directory: t.TempDir(), + Status: StateRunning, + Port: 9100, + } + + _ = pm.Stop("test-agent", info) + + first := <-ch + got, ok := first.Data.(*AgentInfo) + if !ok { + t.Fatalf("event data type = %T, want *AgentInfo", first.Data) + } + if got.Status != StateStopping { + t.Errorf("first event status = %q, want %q", got.Status, StateStopping) + } + if got.Version != "1.2.3" || got.Framework != "langchain" || len(got.Channels) != 1 { + t.Errorf("first event dropped metadata: %+v", got) + } + // The event must be a snapshot, not the live pointer — otherwise + // Stop's later mutations rewrite an already-queued event. + if got == info { + t.Error("event aliases the caller's AgentInfo; broadcast a copy") + } } func TestProcessManagerStopAll(t *testing.T) { diff --git a/forge-ui/static/app.js b/forge-ui/static/app.js index b589331d..f0f5bd78 100644 --- a/forge-ui/static/app.js +++ b/forge-ui/static/app.js @@ -256,12 +256,19 @@ function useSSE(onEvent) { useEffect(() => { const es = new EventSource('/api/events'); - es.addEventListener('agent_status', (e) => { + // EventSource dispatches by `event:` name, so every type the server + // broadcasts needs an explicit listener — an unlisted one is silently + // dropped. agent_created (handlers_create.go) was being dropped, which + // is why a newly created agent only appeared after a refresh or the + // 60s poll. The type is forwarded so the caller can tell them apart. + const forward = (type) => (e) => { try { - const data = JSON.parse(e.data); - callbackRef.current(data); + callbackRef.current(type, JSON.parse(e.data)); } catch { /* ignore parse errors */ } - }); + }; + + es.addEventListener('agent_status', forward('agent_status')); + es.addEventListener('agent_created', forward('agent_created')); es.onerror = () => { // EventSource auto-reconnects @@ -461,6 +468,13 @@ function useChatStream(agentId) { const [sessionId, setSessionId] = useState(null); const abortRef = useRef(null); + // Abort any in-flight stream on unmount. Switching agents remounts this + // hook (ChatPage is keyed by agent id), so without this the previous + // agent's request keeps streaming into a dead component. + useEffect(() => () => { + if (abortRef.current) abortRef.current.abort(); + }, []); + const loadSession = useCallback(async (sid) => { try { const data = await fetchSession(agentId, sid); @@ -3233,7 +3247,15 @@ function App() { }, [loadAgents]); // SSE real-time updates - useSSE((agentData) => { + useSSE((type, agentData) => { + // agent_created carries only {id, directory}, not a full AgentInfo, so + // refetch to pick up the record the cards render from (model, tools, + // channels, status). The merge below deliberately ignores unknown ids, + // so a create can't be handled there. + if (type === 'agent_created') { + loadAgents(); + return; + } setAgents(prev => { const idx = prev.findIndex(a => a.id === agentData.id); if (idx === -1) return prev; @@ -3286,12 +3308,31 @@ function App() { }, [passphrasePrompt]); const handleStop = useCallback(async (id) => { + // Flip to "stopping" on click. The server broadcasts this too, but + // SSEBroker.Broadcast drops events for a full buffer, so the click + // must not depend on the stream to feel responsive. + // + // Capture the prior status so a failure restores what was actually + // there. Hardcoding 'running' would mislabel an agent that was, say, + // already errored — the server's authoritative event reconciles it, + // but not before the card renders the wrong state. Read it from the + // rendered state rather than inside the updater, which is not + // guaranteed to have run by the time the request settles. + const prevStatus = agents.find(a => a.id === id)?.status; + setAgents(prev => prev.map(a => + a.id === id ? { ...a, status: 'stopping', error: '' } : a + )); try { await stopAgent(id); } catch (err) { console.error('Failed to stop agent:', err); + // Mirror ProcessManager.Stop's rollback-to-previous so the card + // doesn't stay stuck on "stopping" with disabled buttons. + setAgents(prev => prev.map(a => + a.id === id ? { ...a, status: prevStatus || 'running', error: err.message } : a + )); } - }, []); + }, [agents]); const handleRescan = useCallback(async () => { setLoading(true); @@ -3310,7 +3351,12 @@ function App() { const renderPage = () => { switch (route.page) { case 'chat': - return html`<${ChatPage} agentId=${route.params.id} agents=${agents} />`; + // key forces a fresh instance per agent. Without it Preact reuses + // the mounted ChatPage on an agentId change, and useState survives + // — so messages/sessionId/streaming stay on the previous agent + // while the header and session list (props / agentId-keyed effect) + // correctly re-render. + return html`<${ChatPage} key=${route.params.id} agentId=${route.params.id} agents=${agents} />`; case 'create': return html`<${CreatePage} />`; case 'config':