diff --git a/apps/connect/internal/hermes/client.go b/apps/connect/internal/hermes/client.go index 37985a5..6c730f4 100644 --- a/apps/connect/internal/hermes/client.go +++ b/apps/connect/internal/hermes/client.go @@ -203,7 +203,7 @@ func legacyCronJobPath(path string) (string, bool) { return "/api/cron/jobs/" + id, true } switch action { - case "pause", "resume", "trigger": + case "pause", "resume", "trigger", "runs": return "/api/cron/jobs/" + id + "/" + action, true default: return "", false diff --git a/apps/connect/internal/hermes/client_test.go b/apps/connect/internal/hermes/client_test.go index 92f4a01..d41f459 100644 --- a/apps/connect/internal/hermes/client_test.go +++ b/apps/connect/internal/hermes/client_test.go @@ -72,6 +72,7 @@ func TestRoutePath(t *testing.T) { {path: "/logs", kind: RouteControlForward, forwardTo: "/api/logs"}, {path: "/jobs/", kind: RouteControlForward, forwardTo: "/api/cron/jobs"}, {path: "/jobs/job_1/pause", kind: RouteControlForward, forwardTo: "/api/cron/jobs/job_1/pause"}, + {path: "/jobs/job_1/runs", kind: RouteControlForward, forwardTo: "/api/cron/jobs/job_1/runs"}, {path: "/jobs/job_1", kind: RouteControlForward, forwardTo: "/api/cron/jobs/job_1"}, {path: "/v1/memory", kind: RouteLocal, localName: "memory"}, {path: "/memory", kind: RouteLocal, localName: "memory"}, diff --git a/apps/connect/internal/hermes/hermes_control.go b/apps/connect/internal/hermes/hermes_control.go index 24ec284..33fef95 100644 --- a/apps/connect/internal/hermes/hermes_control.go +++ b/apps/connect/internal/hermes/hermes_control.go @@ -121,6 +121,8 @@ type controlHeartbeatState struct { Detail string `json:"detail"` LastError string `json:"lastError,omitempty"` RuntimeSessionID string `json:"-"` + OutputSessionID string `json:"outputSessionId,omitempty"` + OutputRuntimeID string `json:"-"` NextAt float64 `json:"-"` InFlight bool `json:"-"` } @@ -443,6 +445,12 @@ func (c *controlClient) SyncHeartbeat(storedSessionID string, runtimeSessionID s } existing, existed := c.heartbeats[storedSessionID] state.RuntimeSessionID = runtimeSessionID + if existed { + // Keep the latest automation run so the scheduler can avoid overlapping + // with it before creating the next isolated heartbeat response session. + state.OutputSessionID = existing.OutputSessionID + state.OutputRuntimeID = existing.OutputRuntimeID + } if existed && !isHeartbeatReplacementCommand(command) && existing.Prompt == state.Prompt && existing.Interval == state.Interval { if existing.FireCount > state.FireCount { state.FireCount = existing.FireCount @@ -548,16 +556,14 @@ func (c *controlClient) fireHeartbeat(storedSessionID string) { ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second) defer cancel() resumed, err := c.Call(ctx, "session.resume", map[string]any{ - "session_id": storedSessionID, - "omit_messages": true, + "session_id": storedSessionID, + "eager_build": true, }) if err != nil { c.finishHeartbeatAttempt(storedSessionID, err, 30*time.Second) return } - var resumedSession struct { - SessionID string `json:"session_id"` - } + var resumedSession heartbeatSourceSession if json.Unmarshal(resumed, &resumedSession) != nil || strings.TrimSpace(resumedSession.SessionID) == "" { c.finishHeartbeatAttempt(storedSessionID, errors.New("Hermes returned an invalid session.resume result for heartbeat"), 30*time.Second) return @@ -577,6 +583,20 @@ func (c *controlClient) fireHeartbeat(storedSessionID string) { c.finishHeartbeatAttempt(storedSessionID, nil, 5*time.Second) return } + outputBusy, err := c.prepareHeartbeatOutput(ctx, state) + if err != nil { + c.finishHeartbeatAttempt(storedSessionID, err, 30*time.Second) + return + } + if outputBusy { + c.finishHeartbeatAttempt(storedSessionID, nil, 5*time.Second) + return + } + outputRuntimeID, outputSessionID, err := c.createHeartbeatOutputSession(ctx, resumedSession) + if err != nil { + c.finishHeartbeatAttempt(storedSessionID, err, 30*time.Second) + return + } pause, err := c.Call(ctx, "slash.exec", map[string]any{ "session_id": runtimeSessionID, "command": "heartbeat pause", @@ -610,8 +630,15 @@ func (c *controlClient) fireHeartbeat(storedSessionID string) { state.Interval, state.Prompt, ) + c.heartbeatMu.Lock() + if current, exists := c.heartbeats[storedSessionID]; exists { + current.OutputRuntimeID = outputRuntimeID + current.OutputSessionID = outputSessionID + c.heartbeats[storedSessionID] = current + } + c.heartbeatMu.Unlock() if _, err = c.Call(ctx, "prompt.submit", map[string]any{ - "session_id": runtimeSessionID, + "session_id": outputRuntimeID, "text": prompt, "queued": true, }); err != nil { @@ -650,6 +677,143 @@ func (c *controlClient) fireHeartbeat(storedSessionID string) { c.heartbeatMu.Unlock() } +type heartbeatSourceSession struct { + SessionID string `json:"session_id"` + Messages []map[string]any `json:"messages"` + Info struct { + CWD string `json:"cwd"` + Model string `json:"model"` + Provider string `json:"provider"` + ReasoningEffort string `json:"reasoning_effort"` + Fast *bool `json:"fast"` + } `json:"info"` +} + +func (c *controlClient) prepareHeartbeatOutput(ctx context.Context, state controlHeartbeatState) (bool, error) { + storedSessionID := strings.TrimSpace(state.OutputSessionID) + if storedSessionID == "" { + return false, nil + } + resumed, err := c.Call(ctx, "session.resume", map[string]any{ + "session_id": storedSessionID, "omit_messages": true, + }) + if err != nil { + var rpcErr *controlRPCError + if errors.As(err, &rpcErr) && rpcErr.Code == 4007 && strings.Contains(strings.ToLower(rpcErr.Message), "not found") { + // The persisted response may have been deleted or pruned. It no + // longer owns a live runtime, so the next firing can replace it. + return false, nil + } + return false, fmt.Errorf("could not inspect the previous heartbeat response: %w", err) + } + var session struct { + SessionID string `json:"session_id"` + } + if json.Unmarshal(resumed, &session) != nil || strings.TrimSpace(session.SessionID) == "" { + return false, errors.New("Hermes returned an invalid heartbeat response session") + } + status, err := c.Call(ctx, "session.status", map[string]any{"session_id": strings.TrimSpace(session.SessionID)}) + if err != nil { + return false, err + } + statusText := strings.ToLower(controlResultOutput(status)) + if statusText == "" { + return false, errors.New("Hermes returned an invalid heartbeat response session status") + } + if strings.Contains(statusText, "agent running: yes") { + return true, nil + } + if _, err := c.Call(ctx, "session.close", map[string]any{"session_id": strings.TrimSpace(session.SessionID)}); err != nil { + return false, fmt.Errorf("could not release the previous heartbeat response runtime: %w", err) + } + return false, nil +} + +func (c *controlClient) createHeartbeatOutputSession(ctx context.Context, source heartbeatSourceSession) (string, string, error) { + params := map[string]any{ + "cols": 100, + "messages": sanitizeHeartbeatSeed(source.Messages), + "source": "heartbeat", + "title": fmt.Sprintf("Heartbeat response %d", time.Now().UnixNano()), + } + if value := strings.TrimSpace(source.Info.CWD); value != "" { + params["cwd"] = value + } + if value := strings.TrimSpace(source.Info.Model); value != "" { + params["model"] = value + } + if value := strings.TrimSpace(source.Info.Provider); value != "" { + params["provider"] = value + } + if value := strings.TrimSpace(source.Info.ReasoningEffort); value != "" { + params["reasoning_effort"] = value + } + if source.Info.Fast != nil { + params["fast"] = *source.Info.Fast + } + created, err := c.Call(ctx, "session.create", params) + if err != nil { + return "", "", fmt.Errorf("could not create the heartbeat response session: %w", err) + } + var session struct { + SessionID string `json:"session_id"` + StoredSessionID string `json:"stored_session_id"` + SessionKey string `json:"session_key"` + } + if json.Unmarshal(created, &session) != nil || strings.TrimSpace(session.SessionID) == "" { + return "", "", errors.New("Hermes returned an invalid session.create result for heartbeat responses") + } + stored := strings.TrimSpace(session.StoredSessionID) + if stored == "" { + stored = strings.TrimSpace(session.SessionKey) + } + if stored == "" { + return "", "", errors.New("Hermes did not return a persisted session id for heartbeat responses") + } + return strings.TrimSpace(session.SessionID), stored, nil +} + +func sanitizeHeartbeatSeed(messages []map[string]any) []map[string]any { + systems := make([]map[string]any, 0) + turns := make([]map[string]any, 0, len(messages)) + pendingUser := "" + for _, message := range messages { + role, _ := message["role"].(string) + role = strings.ToLower(strings.TrimSpace(role)) + content, _ := message["text"].(string) + if strings.TrimSpace(content) == "" { + content, _ = message["content"].(string) + } + content = strings.TrimSpace(content) + if content == "" { + continue + } + switch role { + case "system": + systems = append(systems, map[string]any{"role": "system", "content": content}) + case "user": + if pendingUser == "" { + pendingUser = content + } else { + pendingUser += "\n\n" + content + } + case "assistant": + if pendingUser == "" { + continue + } + turns = append(turns, + map[string]any{"role": "user", "content": pendingUser}, + map[string]any{"role": "assistant", "content": content}, + ) + pendingUser = "" + } + } + // A trailing user turn represents an interrupted/incomplete source turn. + // The heartbeat prompt is itself the next user turn, so do not replay that + // dangling tail or any display-only tool events around it. + return append(systems, turns...) +} + func (c *controlClient) finishHeartbeatAttempt(storedSessionID string, err error, retryAfter time.Duration) { c.heartbeatMu.Lock() defer c.heartbeatMu.Unlock() diff --git a/apps/connect/internal/hermes/hermes_control_test.go b/apps/connect/internal/hermes/hermes_control_test.go index d97edcd..ef96e0f 100644 --- a/apps/connect/internal/hermes/hermes_control_test.go +++ b/apps/connect/internal/hermes/hermes_control_test.go @@ -406,7 +406,32 @@ func TestHeartbeatRunnerQueuesDuePromptAndReanchorsHermes(t *testing.T) { var result any switch request.Method { case "session.resume": - result = map[string]any{"session_id": "runtime-session"} + if request.Params["eager_build"] != true || request.Params["omit_messages"] != nil { + t.Fatalf("source heartbeat resume params = %+v", request.Params) + } + result = map[string]any{ + "session_id": "runtime-session", + "messages": []map[string]any{ + {"role": "user", "text": "Work in this repository"}, + {"role": "assistant", "text": "Understood"}, + }, + "info": map[string]any{ + "cwd": "/tmp/source-workspace", "model": "test-model", "provider": "test-provider", + "reasoning_effort": "high", "fast": false, + }, + } + case "session.create": + title, _ := request.Params["title"].(string) + if request.Params["source"] != "heartbeat" || !strings.HasPrefix(title, "Heartbeat response ") { + t.Fatalf("heartbeat session params = %+v", request.Params) + } + messages, _ := request.Params["messages"].([]any) + if len(messages) != 2 || request.Params["cwd"] != "/tmp/source-workspace" || + request.Params["model"] != "test-model" || request.Params["provider"] != "test-provider" || + request.Params["reasoning_effort"] != "high" || request.Params["fast"] != false { + t.Fatalf("heartbeat did not inherit source context: %+v", request.Params) + } + result = map[string]any{"session_id": "heartbeat-runtime", "stored_session_id": "heartbeat-stored"} case "session.status": result = map[string]any{"output": "Agent Running: No"} case "prompt.submit": @@ -447,7 +472,7 @@ func TestHeartbeatRunnerQueuesDuePromptAndReanchorsHermes(t *testing.T) { ) select { case params := <-promptSubmitted: - if params["queued"] != true || !strings.Contains(params["text"].(string), "[Heartbeat — recurring instruction, fires every 10m]") { + if params["session_id"] != "heartbeat-runtime" || params["queued"] != true || !strings.Contains(params["text"].(string), "[Heartbeat — recurring instruction, fires every 10m]") { t.Fatalf("prompt params = %+v", params) } case <-time.After(3 * time.Second): @@ -456,7 +481,7 @@ func TestHeartbeatRunnerQueuesDuePromptAndReanchorsHermes(t *testing.T) { deadline := time.Now().Add(time.Second) for { state := client.Heartbeat("stored-session") - if state != nil && state.FireCount == 1 && state.NextInSeconds > 500 { + if state != nil && state.FireCount == 1 && state.NextInSeconds > 500 && state.OutputSessionID == "heartbeat-stored" { break } if time.Now().After(deadline) { @@ -466,6 +491,184 @@ func TestHeartbeatRunnerQueuesDuePromptAndReanchorsHermes(t *testing.T) { } } +func TestHeartbeatRunnerDoesNotQueueWhileResponseSessionIsBusy(t *testing.T) { + promptSubmitted := make(chan struct{}, 1) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := websocket.Accept(w, r, nil) + if err != nil { + return + } + defer conn.Close(websocket.StatusNormalClosure, "done") + for { + _, payload, readErr := conn.Read(context.Background()) + if readErr != nil { + return + } + var request struct { + ID string `json:"id"` + Method string `json:"method"` + Params map[string]any `json:"params"` + } + if json.Unmarshal(payload, &request) != nil { + return + } + var result any + switch request.Method { + case "session.resume": + if request.Params["session_id"] == "heartbeat-stored" { + result = map[string]any{"session_id": "heartbeat-runtime", "stored_session_id": "heartbeat-stored"} + } else { + result = map[string]any{"session_id": "runtime-session"} + } + case "session.status": + if request.Params["session_id"] == "heartbeat-runtime" { + result = map[string]any{"output": "Agent Running: Yes"} + } else { + result = map[string]any{"output": "Agent Running: No"} + } + case "prompt.submit": + promptSubmitted <- struct{}{} + result = map[string]any{"status": "streaming"} + default: + t.Fatalf("unexpected method while heartbeat output is busy: %s %+v", request.Method, request.Params) + } + response, _ := json.Marshal(map[string]any{"jsonrpc": "2.0", "id": request.ID, "result": result}) + if conn.Write(context.Background(), websocket.MessageText, response) != nil { + return + } + } + })) + defer server.Close() + + client := newControlClient(Config{ + HermesControlURL: server.URL, + HermesControlToken: "control-token", + }, server.Client()) + defer client.Close() + client.heartbeats["stored-session"] = controlHeartbeatState{ + Status: "active", + Prompt: "Check CI", + Interval: "10m", + OutputSessionID: "heartbeat-stored", + InFlight: true, + } + client.fireHeartbeat("stored-session") + select { + case <-promptSubmitted: + t.Fatal("busy heartbeat response session received another prompt") + default: + } + state := client.Heartbeat("stored-session") + if state == nil || state.InFlight || state.FireCount != 0 || state.NextAt <= float64(time.Now().UnixMilli())/1000 { + t.Fatalf("heartbeat retry state = %+v", state) + } +} + +func TestPrepareHeartbeatOutputAllowsPrunedSession(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := websocket.Accept(w, r, nil) + if err != nil { + return + } + defer conn.Close(websocket.StatusNormalClosure, "done") + _, payload, err := conn.Read(context.Background()) + if err != nil { + return + } + var request struct { + ID string `json:"id"` + } + if json.Unmarshal(payload, &request) != nil { + return + } + response, _ := json.Marshal(map[string]any{ + "jsonrpc": "2.0", "id": request.ID, + "error": map[string]any{"code": 4007, "message": "session not found"}, + }) + _ = conn.Write(context.Background(), websocket.MessageText, response) + })) + defer server.Close() + + client := newControlClient(Config{HermesControlURL: server.URL, HermesControlToken: "control-token"}, server.Client()) + defer client.Close() + busy, err := client.prepareHeartbeatOutput(context.Background(), controlHeartbeatState{OutputSessionID: "pruned-response"}) + if err != nil || busy { + t.Fatalf("pruned heartbeat output blocked the next run: busy=%v err=%v", busy, err) + } +} + +func TestPrepareHeartbeatOutputClosesIdleRuntime(t *testing.T) { + closed := make(chan string, 1) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := websocket.Accept(w, r, nil) + if err != nil { + return + } + defer conn.Close(websocket.StatusNormalClosure, "done") + for { + _, payload, readErr := conn.Read(context.Background()) + if readErr != nil { + return + } + var request struct { + ID string `json:"id"` + Method string `json:"method"` + Params map[string]any `json:"params"` + } + if json.Unmarshal(payload, &request) != nil { + return + } + var result any + switch request.Method { + case "session.resume": + result = map[string]any{"session_id": "idle-runtime"} + case "session.status": + result = map[string]any{"output": "Agent Running: No"} + case "session.close": + closed <- request.Params["session_id"].(string) + result = map[string]any{"closed": true} + default: + t.Fatalf("unexpected method: %s", request.Method) + } + response, _ := json.Marshal(map[string]any{"jsonrpc": "2.0", "id": request.ID, "result": result}) + if conn.Write(context.Background(), websocket.MessageText, response) != nil { + return + } + } + })) + defer server.Close() + + client := newControlClient(Config{HermesControlURL: server.URL, HermesControlToken: "control-token"}, server.Client()) + defer client.Close() + busy, err := client.prepareHeartbeatOutput(context.Background(), controlHeartbeatState{OutputSessionID: "stored-response"}) + if err != nil || busy { + t.Fatalf("idle heartbeat output was not released: busy=%v err=%v", busy, err) + } + select { + case sessionID := <-closed: + if sessionID != "idle-runtime" { + t.Fatalf("closed runtime = %q", sessionID) + } + default: + t.Fatal("idle heartbeat runtime was not closed") + } +} + +func TestSanitizeHeartbeatSeedDropsInterruptedToolTail(t *testing.T) { + seed := sanitizeHeartbeatSeed([]map[string]any{ + {"role": "system", "text": "Use the repository instructions"}, + {"role": "user", "text": "Check the build"}, + {"role": "assistant", "text": "The build is green"}, + {"role": "user", "text": "Now inspect the logs"}, + {"role": "assistant", "text": "", "tool_calls": []any{map[string]any{"name": "terminal"}}}, + {"role": "tool", "text": "unfinished"}, + }) + if len(seed) != 3 || seed[0]["role"] != "system" || seed[1]["role"] != "user" || + seed[2]["role"] != "assistant" || seed[2]["content"] != "The build is green" { + t.Fatalf("sanitized heartbeat seed = %+v", seed) + } +} + func TestControlOperationsAreSerialized(t *testing.T) { client := newControlClient(Config{}, http.DefaultClient) defer client.Close() diff --git a/apps/mobile/src/app/_layout.tsx b/apps/mobile/src/app/_layout.tsx index 7cfdc89..5329214 100644 --- a/apps/mobile/src/app/_layout.tsx +++ b/apps/mobile/src/app/_layout.tsx @@ -139,6 +139,7 @@ export default function RootLayout() { /> + state.connection); + if (!connection) return ; + return ; +} diff --git a/apps/mobile/src/app/explore.tsx b/apps/mobile/src/app/explore.tsx index a8303cd..e87bb30 100644 --- a/apps/mobile/src/app/explore.tsx +++ b/apps/mobile/src/app/explore.tsx @@ -26,7 +26,7 @@ export default function ManageScreen() { }); const sessions = useQuery({ queryKey: ['sessions', connection?.url], - queryFn: () => listSessions(connection!, 5), + queryFn: () => listSessions(connection!, 5, undefined, { excludeSources: ['cron', 'heartbeat'], order: 'recent' }), enabled: Boolean(connection), }); const memory = useQuery({ diff --git a/apps/mobile/src/features/automation/hermes-automation-screen.tsx b/apps/mobile/src/features/automation/hermes-automation-screen.tsx new file mode 100644 index 0000000..e103dba --- /dev/null +++ b/apps/mobile/src/features/automation/hermes-automation-screen.tsx @@ -0,0 +1,374 @@ +import { useQueries, useQuery } from '@tanstack/react-query'; +import { useState } from 'react'; +import { Pressable, RefreshControl, ScrollView, StyleSheet, View } from 'react-native'; +import { SafeAreaView } from 'react-native-safe-area-context'; + +import { AppText, Button, Card, EmptyState, StatusDot } from '@/components/t3-ui'; +import { T3Radius, T3Spacing, T3Typography } from '@/constants/t3-theme'; +import { useT3Theme } from '@/hooks/use-t3-theme'; +import { + getSessionMessages, + listJobRuns, + listJobs, + listSessions, + type AgentConnection, + type HermesJob, + type HermesSession, +} from '@/lib/brio'; +import { + DEFAULT_PROFILE_NAME, + environmentId, + listProfiles, + profileName, +} from '@/lib/profiles'; +import { useProfileStore } from '@/state/profile-store'; + +type AutomationView = 'jobs' | 'heartbeats'; +type AutomationResponse = { + id: string; + content: string; + timestamp: number; +}; + +export function HermesAutomationScreen({ connection }: { connection: AgentConnection }) { + const colors = useT3Theme(); + const [view, setView] = useState('jobs'); + const [selectedJob, setSelectedJob] = useState(null); + const agentId = environmentId(connection); + const storedProfiles = useProfileStore((state) => state.activeProfiles); + const profilesQuery = useQuery({ + queryKey: ['profiles', connection.url, agentId], + queryFn: () => listProfiles(connection), + staleTime: 30_000, + retry: false, + }); + const requestedProfile = storedProfiles[agentId]; + const activeProfile = profilesQuery.data?.profiles.some((profile) => profile.name === requestedProfile) + ? profileName(requestedProfile) + : profilesQuery.data + ? profileName(profilesQuery.data.active) + : DEFAULT_PROFILE_NAME; + + const jobs = useQuery({ + queryKey: ['automation-jobs', connection.id, connection.url, activeProfile], + queryFn: async () => { + const result = await listJobs(connection, activeProfile); + return Array.isArray(result) ? result : (result.jobs ?? []); + }, + refetchInterval: 15_000, + }); + const heartbeatSessions = useQuery({ + queryKey: ['automation-heartbeats', connection.id, connection.url, activeProfile], + queryFn: () => listSessions(connection, 20, activeProfile, { source: 'heartbeat', order: 'recent' }), + refetchInterval: 15_000, + }); + + if (selectedJob) { + return ( + setSelectedJob(null)} + profile={activeProfile} + /> + ); + } + + return ( + + + setView('jobs')} /> + setView('heartbeats')} /> + + {view === 'jobs' ? ( + void jobs.refetch()} />}> + + Scheduled responses + + Hermes runs each job in an isolated session. Brio groups those runs by job and shows the responses here instead of mixing them into chat history. + + + {jobs.isLoading ? : null} + {jobs.isError ? void jobs.refetch()} /> : null} + {(jobs.data ?? []).map((job, index) => ( + setSelectedJob(job)} /> + ))} + {!jobs.isLoading && !jobs.isError && jobs.data?.length === 0 ? ( + + ) : null} + + ) : ( + void heartbeatSessions.refetch()} + refreshing={heartbeatSessions.isRefetching} + oneResponsePerSession + requestPrefix="[Heartbeat — recurring instruction" + /> + )} + + ); +} + +function JobResponses({ + connection, + job, + onBack, + profile, +}: { + connection: AgentConnection; + job: HermesJob; + onBack: () => void; + profile: string; +}) { + const colors = useT3Theme(); + const id = jobID(job); + const runs = useQuery({ + queryKey: ['automation-job-runs', connection.id, connection.url, profile, id], + queryFn: () => listJobRuns(connection, id, 20, profile), + enabled: Boolean(id), + refetchInterval: 15_000, + }); + return ( + + + + + {jobName(job)} + {jobSchedule(job)} + + + void runs.refetch()} + refreshing={runs.isRefetching} + oneResponsePerSession + /> + + ); +} + +function ResponseFeed({ + connection, + detail, + emptyDetail, + emptyTitle, + onRefresh, + oneResponsePerSession = false, + profile, + refreshing, + requestPrefix, + sessions, + sessionsError, + sessionsLoading, + title, +}: { + connection: AgentConnection; + detail: string; + emptyDetail: string; + emptyTitle: string; + onRefresh: () => void; + oneResponsePerSession?: boolean; + profile: string; + refreshing: boolean; + requestPrefix?: string; + sessions: HermesSession[]; + sessionsError: unknown; + sessionsLoading: boolean; + title: string; +}) { + const colors = useT3Theme(); + const responseQueries = useQueries({ + queries: sessions.map((session) => ({ + queryKey: ['automation-response', connection.id, connection.url, profile, session.id, session.message_count], + queryFn: () => getSessionMessages(connection, session.id, profile), + retry: 2, + refetchInterval: (query: { state: { status: string } }) => query.state.status === 'error' ? 30_000 : false, + })), + }); + const successfulSessions: HermesSession[] = []; + const successfulMessages: { role: string; content: string; timestamp: number }[][] = []; + responseQueries.forEach((query, index) => { + if (!query.data) return; + successfulSessions.push(sessions[index]); + successfulMessages.push(query.data.messages); + }); + const responses = collectResponses(successfulSessions, successfulMessages, oneResponsePerSession, requestPrefix); + const responsesLoading = responseQueries.some((query) => query.isLoading); + const responsesRefetching = responseQueries.some((query) => query.isRefetching); + const responseError = responseQueries.find((query) => query.error)?.error; + const refreshAll = () => { + onRefresh(); + responseQueries.forEach((query) => void query.refetch()); + }; + + return ( + }> + + {title} + {detail} + + {sessionsLoading || (sessions.length > 0 && responsesLoading) ? ( + + ) : null} + {sessionsError || responseError ? : null} + {responses.map((response) => )} + {!sessionsLoading && !sessionsError && !responsesLoading && !responseError && responses.length === 0 ? ( + + ) : null} + + ); +} + +function JobCard({ job, onPress }: { job: HermesJob; onPress: () => void }) { + const colors = useT3Theme(); + const paused = job.paused === true || job.enabled === false || job.state === 'paused'; + return ( + ({ opacity: pressed ? 0.6 : 1 })}> + + + + {jobName(job)} + + + {job.prompt || 'Script-only scheduled job'} + + {jobSchedule(job)}{job.last_run_at ? ` · Last response ${formatTime(job.last_run_at)}` : ''} + + + + ); +} + +function ResponseCard({ response }: { response: AutomationResponse }) { + const colors = useT3Theme(); + return ( + + + Hermes + {formatTime(response.timestamp)} + + {response.content} + + ); +} + +function Segment({ active, label, onPress }: { active: boolean; label: string; onPress: () => void }) { + const colors = useT3Theme(); + return ( + + {label} + + ); +} + +function Failure({ error, onRetry }: { error: unknown; onRetry: () => void }) { + return ( + Try again} + detail={error instanceof Error ? error.message : 'The automation response request failed.'} + title="Automation unavailable" + /> + ); +} + +function collectResponses( + sessions: HermesSession[], + messageSets: { role: string; content: string; timestamp: number }[][], + oneResponsePerSession: boolean, + requestPrefix?: string, +) { + const responses: AutomationResponse[] = []; + messageSets.forEach((messages, sessionIndex) => { + let requestIndex = -1; + if (requestPrefix) { + messages.forEach((message, index) => { + if (message.role === 'user' && message.content.trimStart().startsWith(requestPrefix)) { + requestIndex = index; + } + }); + } + if (requestPrefix && requestIndex < 0) return; + const assistant = messages + .slice(requestIndex + 1) + .filter((message) => message.role === 'assistant' && message.content.trim()); + const visible = oneResponsePerSession ? assistant.slice(-1) : assistant; + visible.forEach((message, messageIndex) => { + responses.push({ + id: `${sessions[sessionIndex]?.id ?? sessionIndex}:${message.timestamp}:${messageIndex}`, + content: message.content.trim(), + timestamp: message.timestamp || sessions[sessionIndex]?.started_at || 0, + }); + }); + }); + return responses.sort((left, right) => right.timestamp - left.timestamp); +} + +function jobID(job: HermesJob) { + const id = job.id ?? job.job_id; + return typeof id === 'string' ? id : ''; +} + +function jobName(job: HermesJob) { + return String(job.name || jobID(job) || 'Scheduled job'); +} + +function jobSchedule(job: HermesJob) { + if (job.schedule_display) return job.schedule_display; + if (typeof job.schedule === 'string') return job.schedule; + return job.schedule?.display || job.schedule?.expr || job.schedule?.run_at || 'Scheduled'; +} + +function formatTime(value: string | number) { + const numeric = typeof value === 'number' ? value : Date.parse(value); + const milliseconds = typeof value === 'number' && value < 10_000_000_000 ? value * 1000 : numeric; + if (!Number.isFinite(milliseconds)) return ''; + return new Date(milliseconds).toLocaleString(undefined, { month: 'short', day: 'numeric', hour: 'numeric', minute: '2-digit' }); +} + +const styles = StyleSheet.create({ + safe: { flex: 1 }, + segmented: { alignSelf: 'center', borderRadius: T3Radius.medium, flexDirection: 'row', margin: T3Spacing.lg, maxWidth: 680, padding: 3, width: '90%' }, + segment: { alignItems: 'center', borderRadius: T3Radius.small, flex: 1, minHeight: 40, justifyContent: 'center', paddingHorizontal: T3Spacing.sm }, + segmentLabel: { fontFamily: T3Typography.medium, fontSize: 13, lineHeight: 17 }, + content: { alignSelf: 'center', flexGrow: 1, gap: T3Spacing.md, maxWidth: 720, padding: T3Spacing.xl, paddingBottom: T3Spacing.huge, width: '100%' }, + intro: { gap: T3Spacing.xs, marginBottom: T3Spacing.sm }, + title: { fontFamily: T3Typography.bold, fontSize: 24, lineHeight: 30 }, + detail: { fontSize: 14, lineHeight: 20 }, + jobCard: { gap: T3Spacing.sm, padding: T3Spacing.lg }, + jobTitleRow: { alignItems: 'center', flexDirection: 'row', gap: T3Spacing.sm }, + jobTitle: { flex: 1, fontFamily: T3Typography.bold, fontSize: 16, lineHeight: 21 }, + jobPrompt: { fontSize: 14, lineHeight: 19 }, + meta: { fontSize: 12, lineHeight: 16 }, + detailHeader: { alignItems: 'center', borderBottomWidth: StyleSheet.hairlineWidth, flexDirection: 'row', minHeight: 58, paddingHorizontal: T3Spacing.sm }, + detailHeaderCopy: { flex: 1, paddingRight: 70 }, + detailTitle: { fontFamily: T3Typography.bold, fontSize: 17, lineHeight: 22, textAlign: 'center' }, + responseCard: { gap: T3Spacing.md, padding: T3Spacing.lg }, + responseHeader: { alignItems: 'baseline', flexDirection: 'row', justifyContent: 'space-between' }, + responseLabel: { fontFamily: T3Typography.bold, fontSize: 13, lineHeight: 17 }, + responseText: { fontSize: 15, lineHeight: 22 }, +}); diff --git a/apps/mobile/src/features/command-center/hermes-command-center-screen.tsx b/apps/mobile/src/features/command-center/hermes-command-center-screen.tsx index e498a58..9b49169 100644 --- a/apps/mobile/src/features/command-center/hermes-command-center-screen.tsx +++ b/apps/mobile/src/features/command-center/hermes-command-center-screen.tsx @@ -69,7 +69,13 @@ export function HermesCommandCenterScreen({ connection }: { connection: AgentCon const sessions = useQuery({ queryKey: ['control-sessions', activeConnection.id, activeConnection.url, activeProfile], - queryFn: () => listControlSessions(activeConnection, 100, activeProfile), + queryFn: async () => { + const result = await listControlSessions(activeConnection, 100, activeProfile); + return { + ...result, + sessions: result.sessions.filter((session) => session.source !== 'heartbeat' && session.source !== 'cron'), + }; + }, refetchInterval: 15_000, }); const sessionId = sessions.data?.sessions.some((session) => session.id === selectedSessionId) @@ -356,6 +362,9 @@ export function HermesCommandCenterScreen({ connection }: { connection: AgentCon ) : null} {snapshot.data.heartbeat.prompt} + {snapshot.data.heartbeat.outputSessionId ? ( + + ) : null} {snapshot.data.heartbeat.lastError ? ( Last delivery attempt: {snapshot.data.heartbeat.lastError} diff --git a/apps/mobile/src/features/home/hermes-home-screen.tsx b/apps/mobile/src/features/home/hermes-home-screen.tsx index 8b88d72..9591a9b 100644 --- a/apps/mobile/src/features/home/hermes-home-screen.tsx +++ b/apps/mobile/src/features/home/hermes-home-screen.tsx @@ -128,7 +128,7 @@ export function HermesHomeScreen({ connection }: { connection: AgentConnection } }); const sessions = useQuery({ queryKey: ['sessions', connection.id, connection.url, activeProfile], - queryFn: () => listSessions(connection, 100, activeProfile), + queryFn: () => listSessions(connection, 100, activeProfile, { excludeSources: ['cron', 'heartbeat'], order: 'recent' }), refetchInterval: 15_000, }); const searchResults = useQuery({ @@ -481,6 +481,11 @@ export function HermesHomeScreen({ connection }: { connection: AgentConnection } label="Command Center" onPress={() => openTool('/command-center')} /> + openTool('/automation')} + />
+ router.push('/automation')} /> + setPanel('jobs')} /> setPanel('logs')} /> diff --git a/apps/mobile/src/lib/brio.test.mjs b/apps/mobile/src/lib/brio.test.mjs index b9095bc..9b9f844 100644 --- a/apps/mobile/src/lib/brio.test.mjs +++ b/apps/mobile/src/lib/brio.test.mjs @@ -25,6 +25,9 @@ import { filterAgentsForControlSession, finalizeConnection, getHealth, + listSessions, + listJobRuns, + listJobs, normalizeMessageList, normalizeCapabilities, normalizeFileList, @@ -46,6 +49,110 @@ test('normalizes current Hermes list envelopes without breaking legacy responses assert.deepEqual(normalizeMessageList({ messages }).messages, messages); }); +test('keeps automation sessions out of chat history even when Hermes ignores source filters', async () => { + const originalFetch = globalThis.fetch; + let requestURL = ''; + globalThis.fetch = async (input) => { + requestURL = String(input); + return new Response(JSON.stringify({ + data: [ + { id: 'chat-1', source: 'brio', started_at: 1, message_count: 2 }, + { id: 'cron-1', source: 'cron', started_at: 2, message_count: 2 }, + { id: 'heartbeat-1', source: 'heartbeat', started_at: 3, message_count: 2 }, + ], + }), { status: 200, headers: { 'Content-Type': 'application/json' } }); + }; + try { + const result = await listSessions({ + id: 'direct-1', + name: 'Hermes', + mode: 'self_hosted', + transport: 'direct', + status: 'online', + capabilities: {}, + url: 'http://127.0.0.1:8787', + token: 'secret', + }, 100, 'coder', { excludeSources: ['cron', 'heartbeat'], order: 'recent' }); + assert.deepEqual(result.sessions.map((session) => session.id), ['chat-1']); + assert.match(requestURL, /exclude_sources=cron%2Cheartbeat/); + assert.match(requestURL, /order=recent/); + } finally { + globalThis.fetch = originalFetch; + } +}); + +test('continues through legacy unfiltered pages until it finds visible sessions', async () => { + const originalFetch = globalThis.fetch; + const requestURLs = []; + globalThis.fetch = async (input) => { + const url = new URL(String(input)); + requestURLs.push(url); + const offset = Number(url.searchParams.get('offset') ?? 0); + const data = offset === 0 + ? Array.from({ length: 100 }, (_, index) => ({ + id: `cron-${index}`, + source: index % 2 ? 'cron' : 'heartbeat', + started_at: 200 - index, + message_count: 2, + })) + : [ + { id: 'chat-1', source: 'brio', started_at: 2, message_count: 2 }, + { id: 'chat-2', source: 'cli', started_at: 1, message_count: 2 }, + ]; + return new Response(JSON.stringify({ data }), { + status: 200, + headers: { 'Content-Type': 'application/json' }, + }); + }; + try { + const result = await listSessions({ + id: 'direct-1', + name: 'Hermes', + mode: 'self_hosted', + transport: 'direct', + status: 'online', + capabilities: {}, + url: 'http://127.0.0.1:8787', + token: 'secret', + }, 2, undefined, { excludeSources: ['cron', 'heartbeat'], order: 'recent' }); + assert.deepEqual(result.sessions.map((session) => session.id), ['chat-1', 'chat-2']); + assert.equal(requestURLs.length, 2); + assert.equal(requestURLs[1].searchParams.get('offset'), '100'); + } finally { + globalThis.fetch = originalFetch; + } +}); + +test('scopes default-profile cron lists and run history instead of using Hermes profile=all', async () => { + const originalFetch = globalThis.fetch; + const requests = []; + globalThis.fetch = async (input) => { + requests.push(String(input)); + return new Response( + JSON.stringify(String(input).includes('/runs') ? { runs: [] } : []), + { status: 200, headers: { 'Content-Type': 'application/json' } }, + ); + }; + const connection = { + id: 'direct-1', + name: 'Hermes', + mode: 'self_hosted', + transport: 'direct', + status: 'online', + capabilities: {}, + url: 'http://127.0.0.1:8787', + token: 'secret', + }; + try { + await listJobs(connection, 'default'); + await listJobRuns(connection, 'job-1', 20, 'default'); + assert.equal(requests[0], 'http://127.0.0.1:8787/jobs/?profile=default'); + assert.equal(requests[1], 'http://127.0.0.1:8787/jobs/job-1/runs?limit=20&profile=default'); + } finally { + globalThis.fetch = originalFetch; + } +}); + test('creates a persisted Hermes session before a REST-backed new thread starts', async () => { const originalFetch = globalThis.fetch; let request; diff --git a/apps/mobile/src/lib/brio.ts b/apps/mobile/src/lib/brio.ts index c6b4062..dadfdf3 100644 --- a/apps/mobile/src/lib/brio.ts +++ b/apps/mobile/src/lib/brio.ts @@ -158,6 +158,7 @@ export type HermesSearchResult = { type HermesSessionListEnvelope = { sessions?: HermesSession[]; data?: HermesSession[]; + total?: number; error?: string; }; @@ -295,7 +296,20 @@ export type HermesJob = Record & { prompt?: string; enabled?: boolean; paused?: boolean; - schedule?: string; + state?: string | null; + schedule?: string | { kind?: string; expr?: string; run_at?: string; display?: string }; + schedule_display?: string | null; + last_run_at?: string | null; + next_run_at?: string | null; + last_status?: string | null; + last_error?: string | null; +}; + +export type SessionListOptions = { + source?: string; + sources?: string[]; + excludeSources?: string[]; + order?: 'created' | 'recent'; }; export type HermesControlSession = { @@ -328,6 +342,7 @@ export type HermesHeartbeatStatus = { fireCount: number; detail: string; lastError?: string; + outputSessionId?: string; }; export type HermesSubagent = Record & { @@ -692,12 +707,72 @@ export function interruptComposerSession(connection: AgentConnection, sessionId: }); } -export async function listSessions(connection: AgentConnection, limit = 100, profile?: string) { - const response = await brioFetch( - connection, - `${scopedPath('/api/sessions', profile)}?limit=${limit}`, - ); - return normalizeSessionList(response); +export async function listSessions( + connection: AgentConnection, + limit = 100, + profile?: string, + options: SessionListOptions = {}, +) { + // Older Hermes builds ignore source query parameters. Keep Brio's split + // correct locally while retaining wire compatibility with those builds. If + // a full page is filtered out, continue through bounded offset pages so a + // busy automation history cannot hide later conversations (or vice versa). + const included = options.source + ? new Set([options.source]) + : options.sources?.length + ? new Set(options.sources) + : null; + const excluded = new Set(options.excludeSources ?? []); + const hasSourceFilter = Boolean(included || excluded.size); + const requestedLimit = Math.max(0, limit); + const pageSize = hasSourceFilter ? 100 : Math.min(100, Math.max(1, requestedLimit)); + // Legacy gateways require local filtering. Bound that compatibility scan to + // 1,000 rows; if it is exhausted, fail visibly instead of presenting an + // incorrect empty history or issuing an unbounded burst of requests. + const maxCompatibilityPages = 10; + const sessions: HermesSession[] = []; + const seen = new Set(); + let envelope: HermesSessionListEnvelope = {}; + let offset = 0; + let exhausted = false; + + for (let pageIndex = 0; pageIndex < (hasSourceFilter ? maxCompatibilityPages : 1); pageIndex += 1) { + const query = new URLSearchParams({ limit: String(requestedLimit === 0 ? 0 : pageSize) }); + if (offset) query.set('offset', String(offset)); + if (options.source) query.set('source', options.source); + if (options.sources?.length) query.set('sources', options.sources.join(',')); + if (options.excludeSources?.length) query.set('exclude_sources', options.excludeSources.join(',')); + if (options.order) query.set('order', options.order); + envelope = await brioFetch( + connection, + `${scopedPath('/api/sessions', profile)}?${query.toString()}`, + ); + const page = normalizeSessionList(envelope).sessions; + let sawUnseen = false; + for (const session of page) { + if (seen.has(session.id)) continue; + seen.add(session.id); + sawUnseen = true; + if ((!included || included.has(session.source)) && !excluded.has(session.source)) { + sessions.push(session); + } + } + const reachedReportedTotal = typeof envelope.total === 'number' && offset + pageSize >= envelope.total; + if (sessions.length >= requestedLimit || page.length < pageSize || !sawUnseen || requestedLimit === 0 || reachedReportedTotal) { + exhausted = true; + break; + } + offset += pageSize; + } + + if (hasSourceFilter && !exhausted && sessions.length < requestedLimit) { + throw new Error('Session history is too large to filter safely with this Hermes version. Update Hermes and try again.'); + } + + return { + ...envelope, + sessions: sessions.slice(0, requestedLimit), + }; } export function createSession( @@ -1025,30 +1100,47 @@ export function getLogs( ); } -export function listJobs(connection: AgentConnection) { - return brioFetch(connection, '/jobs/'); +export function listJobs(connection: AgentConnection, profile?: string) { + return brioFetch( + connection, + `${scopedPath('/jobs/', profile)}${defaultAutomationProfileQuery(profile)}`, + ); +} + +export function listJobRuns(connection: AgentConnection, jobId: string, limit = 20, profile?: string) { + const query = new URLSearchParams({ limit: String(Math.max(1, Math.min(limit, 100))) }); + if (!profile || profile === 'default') query.set('profile', 'default'); + return brioFetch<{ runs: HermesSession[]; limit?: number }>( + connection, + `${scopedPath(`/jobs/${encodeURIComponent(jobId)}/runs`, profile)}?${query.toString()}`, + ); } export function runJobAction( connection: AgentConnection, jobId: string, action: 'pause' | 'resume' | 'trigger', + profile?: string, ) { return brioFetch>( connection, - `/jobs/${encodeURIComponent(jobId)}/${action}`, + `${scopedPath(`/jobs/${encodeURIComponent(jobId)}/${action}`, profile)}${defaultAutomationProfileQuery(profile)}`, { method: 'POST', body: '{}' }, ); } -export function deleteJob(connection: AgentConnection, jobId: string) { +export function deleteJob(connection: AgentConnection, jobId: string, profile?: string) { return brioFetch>( connection, - `/jobs/${encodeURIComponent(jobId)}`, + `${scopedPath(`/jobs/${encodeURIComponent(jobId)}`, profile)}${defaultAutomationProfileQuery(profile)}`, { method: 'DELETE' }, ); } +function defaultAutomationProfileQuery(profile?: string) { + return !profile || profile === 'default' ? '?profile=default' : ''; +} + export function controlRPC( connection: AgentConnection, method: string, @@ -1068,8 +1160,12 @@ export function controlRPC( }); } -export function listControlSessions(connection: AgentConnection, limit = 100, profile?: string) { - return controlRPC<{ sessions: HermesControlSession[] }>(connection, 'session.list', { limit }, false, undefined, profile); +export async function listControlSessions(connection: AgentConnection, limit = 100, profile?: string) { + const result = await listSessions(connection, limit, profile, { + excludeSources: ['cron', 'heartbeat', 'kanban', 'tool'], + order: 'recent', + }); + return { sessions: result.sessions as HermesControlSession[] }; } export async function listModelSessions(