feat: human-in-the-loop support with AskUserQuestion

components/backend/handlers/sessions.go

@@ -44,6 +44,9 @@ var (
 	GetGitHubToken                    func(context.Context, kubernetes.Interface, dynamic.Interface, string, string) (string, error)
 	GetGitLabToken                    func(context.Context, kubernetes.Interface, string, string) (string, error)
 	DeriveRepoFolderFromURL           func(string) string
+	// DeriveAgentStatusFromEvents derives agentStatus from the persisted event log.
+	// Set by the websocket package at init to avoid circular imports.
+	DeriveAgentStatusFromEvents func(sessionName string) string
 	// LEGACY: SendMessageToSession removed - AG-UI server uses HTTP/SSE instead of WebSocket
 )
 
@@ -364,6 +367,25 @@ func parseStatus(status map[string]interface{}) *types.AgenticSessionStatus {
 
 // V2 API Handlers - Multi-tenant session management
 
+// enrichAgentStatus derives agentStatus from the persisted event log for
+// Running sessions.  This is the source of truth — it replaces the stale
+// CR-cached value which was subject to goroutine race conditions.
+func enrichAgentStatus(session *types.AgenticSession) {
+	if session.Status == nil || session.Status.Phase != "Running" {
+		return
+	}
+	if DeriveAgentStatusFromEvents == nil {
+		return
+	}
+	name, _ := session.Metadata["name"].(string)
+	if name == "" {
+		return
+	}
+	if derived := DeriveAgentStatusFromEvents(name); derived != "" {
+		session.Status.AgentStatus = types.StringPtr(derived)
+	}
+}
+
 func ListSessions(c *gin.Context) {
 	project := c.GetString("project")
 
@@ -447,6 +469,11 @@ func ListSessions(c *gin.Context) {
 	totalCount := len(sessions)
 	paginatedSessions, hasMore, nextOffset := paginateSessions(sessions, params.Offset, params.Limit)
 
+	// Derive agentStatus from event log only for paginated sessions (performance optimization)
+	for i := range paginatedSessions {
+		enrichAgentStatus(&paginatedSessions[i])
+	}
+
 	response := types.PaginatedResponse{
 		Items:      paginatedSessions,
 		TotalCount: totalCount,
@@ -919,6 +946,9 @@ func GetSession(c *gin.Context) {
 		session.Status = parseStatus(status)
 	}
 
+	// Derive agentStatus from event log (source of truth) for running sessions
+	enrichAgentStatus(&session)
+
 	session.AutoBranch = ComputeAutoBranch(sessionName)
 
 	c.JSON(http.StatusOK, session)

components/backend/main.go

@@ -166,6 +166,7 @@ func main() {
 
 	// Initialize websocket package
 	websocket.StateBaseDir = server.StateBaseDir
+	handlers.DeriveAgentStatusFromEvents = websocket.DeriveAgentStatus
 
 	// Normal server mode
 	if err := server.Run(registerRoutes); err != nil {

components/backend/types/agui.go

@@ -69,6 +69,13 @@ const (
 	EventTypeMeta = "META"
 )
 
+// Agent status values derived from the AG-UI event stream.
+const (
+	AgentStatusWorking      = "working"
+	AgentStatusIdle         = "idle"
+	AgentStatusWaitingInput = "waiting_input"
+)
+
 // AG-UI Message Roles
 // See: https://docs.ag-ui.com/concepts/messages
 const (

components/backend/types/session.go

@@ -42,6 +42,7 @@ type AgenticSessionStatus struct {
 	StartTime          *string             `json:"startTime,omitempty"`
 	CompletionTime     *string             `json:"completionTime,omitempty"`
 	LastActivityTime   *string             `json:"lastActivityTime,omitempty"`
+	AgentStatus        *string             `json:"agentStatus,omitempty"`
 	StoppedReason      *string             `json:"stoppedReason,omitempty"`
 	ReconciledRepos    []ReconciledRepo    `json:"reconciledRepos,omitempty"`
 	ReconciledWorkflow *ReconciledWorkflow `json:"reconciledWorkflow,omitempty"`

components/backend/websocket/agui_proxy.go

@@ -439,15 +439,19 @@ func persistStreamedEvent(sessionID, runID, threadID, jsonData string) {
 
 	persistEvent(sessionID, event)
 
-	// Update lastActivityTime on CR for activity events (debounced).
-	// Extract event type to check; projectName is derived from the
+	// Extract event type; projectName is derived from the
 	// sessionID-to-project mapping populated by HandleAGUIRunProxy.
 	eventType, _ := event["type"].(string)
+
+	// Update lastActivityTime on CR for activity events (debounced).
 	if isActivityEvent(eventType) {
 		if projectName, ok := sessionProjectMap.Load(sessionID); ok {
 			updateLastActivityTime(projectName.(string), sessionID, eventType == types.EventTypeRunStarted)
 		}
 	}
+
+	// agentStatus is derived at query time from the event log (DeriveAgentStatus).
+	// No CR updates needed here — the persisted events ARE the source of truth.
 }
 
 // ─── POST /agui/interrupt ────────────────────────────────────────────
@@ -948,3 +952,16 @@ func updateLastActivityTime(projectName, sessionName string, immediate bool) {
 		}
 	}()
 }
+
+// isAskUserQuestionToolCall checks if a tool call name is the AskUserQuestion HITL tool.
+// Uses case-insensitive comparison after stripping non-alpha characters,
+// matching the frontend pattern in use-agent-status.ts.
+func isAskUserQuestionToolCall(name string) bool {
+	var clean strings.Builder
+	for _, r := range strings.ToLower(name) {
+		if r >= 'a' && r <= 'z' {
+			clean.WriteRune(r)
+		}
+	}
+	return clean.String() == "askuserquestion"
+}

components/backend/websocket/agui_store.go

@@ -12,6 +12,7 @@ package websocket
 import (
 	"ambient-code-backend/types"
 	"bufio"
+	"bytes"
 	"encoding/json"
 	"fmt"
 	"log"
@@ -210,6 +211,108 @@ func loadEvents(sessionID string) []map[string]interface{} {
 	return events
 }
 
+// DeriveAgentStatus reads a session's event log and returns the agent
+// status derived from the last significant events.
+//
+// Returns "" if the status cannot be determined (no events, file missing, etc.).
+func DeriveAgentStatus(sessionID string) string {
+	path := fmt.Sprintf("%s/sessions/%s/agui-events.jsonl", StateBaseDir, sessionID)
+
+	// Read only the tail of the file to avoid loading entire event log into memory.
+	// 64KB is sufficient for recent lifecycle events (scanning backwards).
+	const maxTailBytes = 64 * 1024
+
+	file, err := os.Open(path)
+	if err != nil {
+		return ""
+	}
+	defer file.Close()
+
+	stat, err := file.Stat()
+	if err != nil {
+		return ""
+	}
+
+	fileSize := stat.Size()
+	var data []byte
+
+	if fileSize <= maxTailBytes {
+		// File is small, read it all
+		data, err = os.ReadFile(path)
+		if err != nil {
+			return ""
+		}
+	} else {
+		// File is large, seek to tail and read last N bytes
+		offset := fileSize - maxTailBytes
+		_, err = file.Seek(offset, 0)
+		if err != nil {
+			return ""
+		}
+
+		data = make([]byte, maxTailBytes)
+		n, err := file.Read(data)
+		if err != nil {
+			return ""
+		}
+		data = data[:n]
+
+		// Skip partial first line (we seeked into the middle of a line)
+		if idx := bytes.IndexByte(data, '\n'); idx >= 0 {
+			data = data[idx+1:]
+		}
+	}
+
+	lines := splitLines(data)
+
+	// Scan backwards.  We only care about lifecycle and AskUserQuestion events.
+	//   RUN_STARTED                       → "working"
+	//   RUN_FINISHED / RUN_ERROR          → "idle", unless same run had AskUserQuestion
+	//   TOOL_CALL_START (AskUserQuestion) → "waiting_input"
+	var runEndRunID string // set when we hit RUN_FINISHED/RUN_ERROR and need to look deeper
+	for i := len(lines) - 1; i >= 0; i-- {
+		if len(lines[i]) == 0 {
+			continue
+		}
+		var evt map[string]interface{}
+		if err := json.Unmarshal(lines[i], &evt); err != nil {
+			continue
+		}
+		evtType, _ := evt["type"].(string)
+
+		switch evtType {
+		case types.EventTypeRunStarted:
+			if runEndRunID != "" {
+				// We were scanning for an AskUserQuestion but hit RUN_STARTED first → idle
+				return types.AgentStatusIdle
+			}
+			return types.AgentStatusWorking
+
+		case types.EventTypeRunFinished, types.EventTypeRunError:
+			if runEndRunID == "" {
+				// First run-end seen; scan deeper within this run for AskUserQuestion
+				runEndRunID, _ = evt["runId"].(string)
+			}
+
+		case types.EventTypeToolCallStart:
+			if runEndRunID != "" {
+				// Only relevant if we're scanning within the ended run
+				if evtRunID, _ := evt["runId"].(string); evtRunID != "" && evtRunID != runEndRunID {
+					return types.AgentStatusIdle
+				}
+			}
+			if toolName, _ := evt["toolCallName"].(string); isAskUserQuestionToolCall(toolName) {
+				return types.AgentStatusWaitingInput
+			}
+		}
+	}
+
+	if runEndRunID != "" {
+		return types.AgentStatusIdle
+	}
+	return ""
+}
+
 // ─── Compaction ──────────────────────────────────────────────────────
 //
 // Go port of @ag-ui/client compactEvents.  Concatenates streaming deltas