Skip to content

Commit 66c12a2

Browse files
Gkrumbach07claudeAmbient Code Botambient-code[bot]
committed
feat: human-in-the-loop support with AskUserQuestion (#871)
## Summary - **AskUserQuestion UI**: Interactive question component with single/multi-select options, freeform "Other" input, and multi-question tabbed navigation - **Agent status from event log**: Derive `agentStatus` at query time from persisted AG-UI events instead of caching on the CR (eliminates goroutine race conditions) - **Frontend status fix**: `useAgentStatus` hook reads raw `PlatformMessage.toolCalls[]` format, correctly detecting unanswered AskUserQuestion tool calls - **TOOL_CALL_RESULT emission**: Adapter emits `TOOL_CALL_RESULT` on next run for halted tool calls so the frontend transitions questions to answered state - **Session status indicators**: `SessionStatusDot` and `AgentStatusIndicator` components for detail and table views - **CRD cleanup**: Removed `agentStatus` field from CRD schema — no longer stored, only derived Supersedes #725. ## Test plan - [ ] Create session, send prompt that triggers AskUserQuestion - [ ] Verify question UI appears with options - [ ] Verify session table shows `waiting_input` status - [ ] Verify detail page shows `waiting_input` status - [ ] Submit answer, verify question transitions to answered (green) state - [ ] Verify agent resumes after answer 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com> Co-authored-by: Ambient Code Bot <bot@ambient-code.local> Co-authored-by: ambient-code[bot] <ambient-code[bot]@users.noreply.github.com>
1 parent 373db39 commit 66c12a2

24 files changed

Lines changed: 1239 additions & 27 deletions

File tree

components/backend/handlers/sessions.go

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,9 @@ var (
4141
GetGitHubToken func(context.Context, kubernetes.Interface, dynamic.Interface, string, string) (string, error)
4242
GetGitLabToken func(context.Context, kubernetes.Interface, string, string) (string, error)
4343
DeriveRepoFolderFromURL func(string) string
44+
// DeriveAgentStatusFromEvents derives agentStatus from the persisted event log.
45+
// Set by the websocket package at init to avoid circular imports.
46+
DeriveAgentStatusFromEvents func(sessionName string) string
4447
// LEGACY: SendMessageToSession removed - AG-UI server uses HTTP/SSE instead of WebSocket
4548
)
4649

@@ -361,6 +364,25 @@ func parseStatus(status map[string]interface{}) *types.AgenticSessionStatus {
361364

362365
// V2 API Handlers - Multi-tenant session management
363366

367+
// enrichAgentStatus derives agentStatus from the persisted event log for
368+
// Running sessions. This is the source of truth — it replaces the stale
369+
// CR-cached value which was subject to goroutine race conditions.
370+
func enrichAgentStatus(session *types.AgenticSession) {
371+
if session.Status == nil || session.Status.Phase != "Running" {
372+
return
373+
}
374+
if DeriveAgentStatusFromEvents == nil {
375+
return
376+
}
377+
name, _ := session.Metadata["name"].(string)
378+
if name == "" {
379+
return
380+
}
381+
if derived := DeriveAgentStatusFromEvents(name); derived != "" {
382+
session.Status.AgentStatus = types.StringPtr(derived)
383+
}
384+
}
385+
364386
func ListSessions(c *gin.Context) {
365387
project := c.GetString("project")
366388

@@ -431,6 +453,11 @@ func ListSessions(c *gin.Context) {
431453
totalCount := len(sessions)
432454
paginatedSessions, hasMore, nextOffset := paginateSessions(sessions, params.Offset, params.Limit)
433455

456+
// Derive agentStatus from event log only for paginated sessions (performance optimization)
457+
for i := range paginatedSessions {
458+
enrichAgentStatus(&paginatedSessions[i])
459+
}
460+
434461
response := types.PaginatedResponse{
435462
Items: paginatedSessions,
436463
TotalCount: totalCount,
@@ -903,6 +930,9 @@ func GetSession(c *gin.Context) {
903930
session.Status = parseStatus(status)
904931
}
905932

933+
// Derive agentStatus from event log (source of truth) for running sessions
934+
enrichAgentStatus(&session)
935+
906936
session.AutoBranch = ComputeAutoBranch(sessionName)
907937

908938
c.JSON(http.StatusOK, session)

components/backend/main.go

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -166,6 +166,7 @@ func main() {
166166

167167
// Initialize websocket package
168168
websocket.StateBaseDir = server.StateBaseDir
169+
handlers.DeriveAgentStatusFromEvents = websocket.DeriveAgentStatus
169170

170171
// Normal server mode
171172
if err := server.Run(registerRoutes); err != nil {

components/backend/types/agui.go

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -69,6 +69,13 @@ const (
6969
EventTypeMeta = "META"
7070
)
7171

72+
// Agent status values derived from the AG-UI event stream.
73+
const (
74+
AgentStatusWorking = "working"
75+
AgentStatusIdle = "idle"
76+
AgentStatusWaitingInput = "waiting_input"
77+
)
78+
7279
// AG-UI Message Roles
7380
// See: https://docs.ag-ui.com/concepts/messages
7481
const (

components/backend/types/session.go

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@ type AgenticSessionStatus struct {
4242
StartTime *string `json:"startTime,omitempty"`
4343
CompletionTime *string `json:"completionTime,omitempty"`
4444
LastActivityTime *string `json:"lastActivityTime,omitempty"`
45+
AgentStatus *string `json:"agentStatus,omitempty"`
4546
StoppedReason *string `json:"stoppedReason,omitempty"`
4647
ReconciledRepos []ReconciledRepo `json:"reconciledRepos,omitempty"`
4748
ReconciledWorkflow *ReconciledWorkflow `json:"reconciledWorkflow,omitempty"`

components/backend/websocket/agui_proxy.go

Lines changed: 19 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -436,15 +436,19 @@ func persistStreamedEvent(sessionID, runID, threadID, jsonData string) {
436436

437437
persistEvent(sessionID, event)
438438

439-
// Update lastActivityTime on CR for activity events (debounced).
440-
// Extract event type to check; projectName is derived from the
439+
// Extract event type; projectName is derived from the
441440
// sessionID-to-project mapping populated by HandleAGUIRunProxy.
442441
eventType, _ := event["type"].(string)
442+
443+
// Update lastActivityTime on CR for activity events (debounced).
443444
if isActivityEvent(eventType) {
444445
if projectName, ok := sessionProjectMap.Load(sessionID); ok {
445446
updateLastActivityTime(projectName.(string), sessionID, eventType == types.EventTypeRunStarted)
446447
}
447448
}
449+
450+
// agentStatus is derived at query time from the event log (DeriveAgentStatus).
451+
// No CR updates needed here — the persisted events ARE the source of truth.
448452
}
449453

450454
// ─── POST /agui/interrupt ────────────────────────────────────────────
@@ -945,3 +949,16 @@ func updateLastActivityTime(projectName, sessionName string, immediate bool) {
945949
}
946950
}()
947951
}
952+
953+
// isAskUserQuestionToolCall checks if a tool call name is the AskUserQuestion HITL tool.
954+
// Uses case-insensitive comparison after stripping non-alpha characters,
955+
// matching the frontend pattern in use-agent-status.ts.
956+
func isAskUserQuestionToolCall(name string) bool {
957+
var clean strings.Builder
958+
for _, r := range strings.ToLower(name) {
959+
if r >= 'a' && r <= 'z' {
960+
clean.WriteRune(r)
961+
}
962+
}
963+
return clean.String() == "askuserquestion"
964+
}

components/backend/websocket/agui_store.go

Lines changed: 103 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ package websocket
1111

1212
import (
1313
"ambient-code-backend/types"
14+
"bytes"
1415
"encoding/json"
1516
"fmt"
1617
"log"
@@ -194,6 +195,108 @@ func loadEvents(sessionID string) []map[string]interface{} {
194195
return events
195196
}
196197

198+
// DeriveAgentStatus reads a session's event log and returns the agent
199+
// status derived from the last significant events.
200+
//
201+
// Returns "" if the status cannot be determined (no events, file missing, etc.).
202+
func DeriveAgentStatus(sessionID string) string {
203+
path := fmt.Sprintf("%s/sessions/%s/agui-events.jsonl", StateBaseDir, sessionID)
204+
205+
// Read only the tail of the file to avoid loading entire event log into memory.
206+
// 64KB is sufficient for recent lifecycle events (scanning backwards).
207+
const maxTailBytes = 64 * 1024
208+
209+
file, err := os.Open(path)
210+
if err != nil {
211+
return ""
212+
}
213+
defer file.Close()
214+
215+
stat, err := file.Stat()
216+
if err != nil {
217+
return ""
218+
}
219+
220+
fileSize := stat.Size()
221+
var data []byte
222+
223+
if fileSize <= maxTailBytes {
224+
// File is small, read it all
225+
data, err = os.ReadFile(path)
226+
if err != nil {
227+
return ""
228+
}
229+
} else {
230+
// File is large, seek to tail and read last N bytes
231+
offset := fileSize - maxTailBytes
232+
_, err = file.Seek(offset, 0)
233+
if err != nil {
234+
return ""
235+
}
236+
237+
data = make([]byte, maxTailBytes)
238+
n, err := file.Read(data)
239+
if err != nil {
240+
return ""
241+
}
242+
data = data[:n]
243+
244+
// Skip partial first line (we seeked into the middle of a line)
245+
if idx := bytes.IndexByte(data, '\n'); idx >= 0 {
246+
data = data[idx+1:]
247+
}
248+
}
249+
250+
lines := splitLines(data)
251+
252+
// Scan backwards. We only care about lifecycle and AskUserQuestion events.
253+
// RUN_STARTED → "working"
254+
// RUN_FINISHED / RUN_ERROR → "idle", unless same run had AskUserQuestion
255+
// TOOL_CALL_START (AskUserQuestion) → "waiting_input"
256+
var runEndRunID string // set when we hit RUN_FINISHED/RUN_ERROR and need to look deeper
257+
for i := len(lines) - 1; i >= 0; i-- {
258+
if len(lines[i]) == 0 {
259+
continue
260+
}
261+
var evt map[string]interface{}
262+
if err := json.Unmarshal(lines[i], &evt); err != nil {
263+
continue
264+
}
265+
evtType, _ := evt["type"].(string)
266+
267+
switch evtType {
268+
case types.EventTypeRunStarted:
269+
if runEndRunID != "" {
270+
// We were scanning for an AskUserQuestion but hit RUN_STARTED first → idle
271+
return types.AgentStatusIdle
272+
}
273+
return types.AgentStatusWorking
274+
275+
case types.EventTypeRunFinished, types.EventTypeRunError:
276+
if runEndRunID == "" {
277+
// First run-end seen; scan deeper within this run for AskUserQuestion
278+
runEndRunID, _ = evt["runId"].(string)
279+
}
280+
281+
case types.EventTypeToolCallStart:
282+
if runEndRunID != "" {
283+
// Only relevant if we're scanning within the ended run
284+
if evtRunID, _ := evt["runId"].(string); evtRunID != "" && evtRunID != runEndRunID {
285+
return types.AgentStatusIdle
286+
}
287+
}
288+
if toolName, _ := evt["toolCallName"].(string); isAskUserQuestionToolCall(toolName) {
289+
return types.AgentStatusWaitingInput
290+
}
291+
}
292+
}
293+
294+
if runEndRunID != "" {
295+
return types.AgentStatusIdle
296+
}
297+
return ""
298+
}
299+
197300
// ─── Compaction ──────────────────────────────────────────────────────
198301
//
199302
// Go port of @ag-ui/client compactEvents. Concatenates streaming deltas

0 commit comments

Comments
 (0)