Skip to content

Commit c493dec

Browse files
Gkrumbach07claudeAmbient Code Botambient-code[bot]
committed
feat: human-in-the-loop support with AskUserQuestion (#871)
- **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. - [ ] 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 2f0b13d commit c493dec

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
@@ -44,6 +44,9 @@ var (
4444
GetGitHubToken func(context.Context, kubernetes.Interface, dynamic.Interface, string, string) (string, error)
4545
GetGitLabToken func(context.Context, kubernetes.Interface, string, string) (string, error)
4646
DeriveRepoFolderFromURL func(string) string
47+
// DeriveAgentStatusFromEvents derives agentStatus from the persisted event log.
48+
// Set by the websocket package at init to avoid circular imports.
49+
DeriveAgentStatusFromEvents func(sessionName string) string
4750
// LEGACY: SendMessageToSession removed - AG-UI server uses HTTP/SSE instead of WebSocket
4851
)
4952

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

365368
// V2 API Handlers - Multi-tenant session management
366369

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

@@ -447,6 +469,11 @@ func ListSessions(c *gin.Context) {
447469
totalCount := len(sessions)
448470
paginatedSessions, hasMore, nextOffset := paginateSessions(sessions, params.Offset, params.Limit)
449471

472+
// Derive agentStatus from event log only for paginated sessions (performance optimization)
473+
for i := range paginatedSessions {
474+
enrichAgentStatus(&paginatedSessions[i])
475+
}
476+
450477
response := types.PaginatedResponse{
451478
Items: paginatedSessions,
452479
TotalCount: totalCount,
@@ -919,6 +946,9 @@ func GetSession(c *gin.Context) {
919946
session.Status = parseStatus(status)
920947
}
921948

949+
// Derive agentStatus from event log (source of truth) for running sessions
950+
enrichAgentStatus(&session)
951+
922952
session.AutoBranch = ComputeAutoBranch(sessionName)
923953

924954
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
@@ -439,15 +439,19 @@ func persistStreamedEvent(sessionID, runID, threadID, jsonData string) {
439439

440440
persistEvent(sessionID, event)
441441

442-
// Update lastActivityTime on CR for activity events (debounced).
443-
// Extract event type to check; projectName is derived from the
442+
// Extract event type; projectName is derived from the
444443
// sessionID-to-project mapping populated by HandleAGUIRunProxy.
445444
eventType, _ := event["type"].(string)
445+
446+
// Update lastActivityTime on CR for activity events (debounced).
446447
if isActivityEvent(eventType) {
447448
if projectName, ok := sessionProjectMap.Load(sessionID); ok {
448449
updateLastActivityTime(projectName.(string), sessionID, eventType == types.EventTypeRunStarted)
449450
}
450451
}
452+
453+
// agentStatus is derived at query time from the event log (DeriveAgentStatus).
454+
// No CR updates needed here — the persisted events ARE the source of truth.
451455
}
452456

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

components/backend/websocket/agui_store.go

Lines changed: 103 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ package websocket
1212
import (
1313
"ambient-code-backend/types"
1414
"bufio"
15+
"bytes"
1516
"encoding/json"
1617
"fmt"
1718
"log"
@@ -210,6 +211,108 @@ func loadEvents(sessionID string) []map[string]interface{} {
210211
return events
211212
}
212213

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

0 commit comments

Comments
 (0)