From 0d8195dddc8ce1d52b66a97700d2789ce2c91fb8 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Fri, 10 Jul 2026 15:29:09 -0400 Subject: [PATCH 1/9] feat(control-plane): authenticated approval resolution endpoint by execution ID Approvals could previously only be resolved through the HMAC-signed webhook (POST /api/v1/webhooks/approval-response), whose secret is held by the external approval service. Extract the resolution core (idempotency, state transitions, approval bookkeeping, events, agent callback) into webhookApprovalController.applyApprovalDecision and add POST /api/v1/executions/:execution_id/approval-response under the authenticated agentAPI group, so operators and CLIs holding an API key can approve/reject/request_changes directly. Webhook behavior is unchanged; 'expired' stays webhook-only since expiry is not an operator decision. Co-Authored-By: Claude Fable 5 --- .../handlers/execution_approval_response.go | 87 +++++++ .../execution_approval_response_test.go | 213 ++++++++++++++++++ .../internal/handlers/webhook_approval.go | 17 +- control-plane/internal/server/routes_core.go | 4 + 4 files changed, 318 insertions(+), 3 deletions(-) create mode 100644 control-plane/internal/handlers/execution_approval_response.go create mode 100644 control-plane/internal/handlers/execution_approval_response_test.go diff --git a/control-plane/internal/handlers/execution_approval_response.go b/control-plane/internal/handlers/execution_approval_response.go new file mode 100644 index 000000000..ee2101003 --- /dev/null +++ b/control-plane/internal/handlers/execution_approval_response.go @@ -0,0 +1,87 @@ +package handlers + +import ( + "encoding/json" + "fmt" + "net/http" + "strings" + + "github.com/Agent-Field/agentfield/control-plane/internal/logger" + + "github.com/gin-gonic/gin" +) + +// ResolveApprovalRequest is the body for POST /executions/:execution_id/approval-response. +type ResolveApprovalRequest struct { + Decision string `json:"decision" binding:"required"` + Reason string `json:"reason,omitempty"` + Response json.RawMessage `json:"response,omitempty"` +} + +// ResolveApprovalHandler resolves a pending approval by execution ID under the +// standard API auth middleware. The HMAC-signed webhook +// (/api/v1/webhooks/approval-response) is keyed by approval_request_id and +// requires the shared webhook secret held by the external approval service; +// this endpoint lets an authenticated operator (e.g. the CLI) resolve the same +// approval directly. Both paths share applyApprovalDecision, so state +// transitions, idempotency, events, and callbacks are identical. +func ResolveApprovalHandler(store ExecutionStore) gin.HandlerFunc { + ctrl := &webhookApprovalController{store: store} + return ctrl.handleResolveApproval +} + +func (c *webhookApprovalController) handleResolveApproval(ctx *gin.Context) { + executionID := ctx.Param("execution_id") + if executionID == "" { + ctx.JSON(http.StatusBadRequest, gin.H{"error": "execution_id is required"}) + return + } + + var req ResolveApprovalRequest + if err := ctx.ShouldBindJSON(&req); err != nil { + ctx.JSON(http.StatusBadRequest, gin.H{"error": fmt.Sprintf("invalid request body: %v", err)}) + return + } + + // "expired" is deliberately excluded: expiry is an approval-service + // outcome, not an operator decision. + decision := normalizeDecision(strings.TrimSpace(req.Decision)) + switch decision { + case "approved", "rejected", "request_changes": + // valid + default: + ctx.JSON(http.StatusBadRequest, gin.H{ + "error": "invalid_decision", + "message": fmt.Sprintf("invalid decision '%s'; must be approved, rejected, or request_changes", req.Decision), + }) + return + } + + reqCtx := ctx.Request.Context() + wfExec, err := c.store.GetWorkflowExecution(reqCtx, executionID) + if err != nil { + logger.Logger.Error().Err(err).Str("execution_id", executionID).Msg("failed to get workflow execution for approval resolution") + ctx.JSON(http.StatusInternalServerError, gin.H{"error": "failed to look up execution"}) + return + } + if wfExec == nil { + ctx.JSON(http.StatusNotFound, gin.H{"error": fmt.Sprintf("execution %s not found", executionID)}) + return + } + if wfExec.ApprovalRequestID == nil || *wfExec.ApprovalRequestID == "" { + ctx.JSON(http.StatusNotFound, gin.H{ + "error": "no_approval_request", + "message": "No approval request exists for this execution", + }) + return + } + + payload := &ApprovalWebhookPayload{ + RequestID: *wfExec.ApprovalRequestID, + Decision: decision, + Feedback: req.Reason, + Response: req.Response, + } + + c.applyApprovalDecision(ctx, executionID, wfExec, payload, decision) +} diff --git a/control-plane/internal/handlers/execution_approval_response_test.go b/control-plane/internal/handlers/execution_approval_response_test.go new file mode 100644 index 000000000..660e0272f --- /dev/null +++ b/control-plane/internal/handlers/execution_approval_response_test.go @@ -0,0 +1,213 @@ +package handlers + +import ( + "bytes" + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/Agent-Field/agentfield/control-plane/pkg/types" + + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func resolveApprovalRouter(store ExecutionStore) *gin.Engine { + gin.SetMode(gin.TestMode) + router := gin.New() + router.POST("/api/v1/executions/:execution_id/approval-response", ResolveApprovalHandler(store)) + return router +} + +func postResolveApproval(t *testing.T, router *gin.Engine, executionID string, body map[string]any) *httptest.ResponseRecorder { + t.Helper() + + encoded, err := json.Marshal(body) + require.NoError(t, err) + + req := httptest.NewRequest(http.MethodPost, "/api/v1/executions/"+executionID+"/approval-response", bytes.NewReader(encoded)) + req.Header.Set("Content-Type", "application/json") + resp := httptest.NewRecorder() + router.ServeHTTP(resp, req) + return resp +} + +// Contract: approving a waiting execution transitions it back to running and +// records the decision, exactly like the HMAC webhook path. +func TestResolveApproval_Approved(t *testing.T) { + store := seedWaitingExecution(t, "exec-1", "agent-1", "req-abc") + router := resolveApprovalRouter(store) + + resp := postResolveApproval(t, router, "exec-1", map[string]any{ + "decision": "approved", + "reason": "Looks good!", + }) + + require.Equal(t, http.StatusOK, resp.Code) + + var result map[string]any + require.NoError(t, json.Unmarshal(resp.Body.Bytes(), &result)) + assert.Equal(t, "processed", result["status"]) + assert.Equal(t, "exec-1", result["execution_id"]) + assert.Equal(t, "approved", result["decision"]) + assert.Equal(t, types.ExecutionStatusRunning, result["new_status"]) + + wfExec, err := store.GetWorkflowExecution(context.Background(), "exec-1") + require.NoError(t, err) + assert.Equal(t, types.ExecutionStatusRunning, wfExec.Status) + require.NotNil(t, wfExec.ApprovalStatus) + assert.Equal(t, "approved", *wfExec.ApprovalStatus) + assert.NotNil(t, wfExec.ApprovalRespondedAt) + require.NotNil(t, wfExec.ApprovalResponse) + assert.Contains(t, *wfExec.ApprovalResponse, "Looks good!") + + execRecord, err := store.GetExecutionRecord(context.Background(), "exec-1") + require.NoError(t, err) + assert.Equal(t, types.ExecutionStatusRunning, execRecord.Status) +} + +// Contract: rejecting cancels the execution and stamps completion. +func TestResolveApproval_Rejected(t *testing.T) { + store := seedWaitingExecution(t, "exec-1", "agent-1", "req-abc") + router := resolveApprovalRouter(store) + + resp := postResolveApproval(t, router, "exec-1", map[string]any{ + "decision": "rejected", + "reason": "not safe", + }) + + require.Equal(t, http.StatusOK, resp.Code) + + var result map[string]any + require.NoError(t, json.Unmarshal(resp.Body.Bytes(), &result)) + assert.Equal(t, "processed", result["status"]) + assert.Equal(t, types.ExecutionStatusCancelled, result["new_status"]) + + wfExec, err := store.GetWorkflowExecution(context.Background(), "exec-1") + require.NoError(t, err) + assert.Equal(t, types.ExecutionStatusCancelled, wfExec.Status) + assert.NotNil(t, wfExec.CompletedAt) + require.NotNil(t, wfExec.StatusReason) + assert.Contains(t, *wfExec.StatusReason, "approval_rejected") + assert.Contains(t, *wfExec.StatusReason, "not safe") +} + +// Contract: decision synonyms accepted by the webhook normalize here too. +func TestResolveApproval_NormalizesDecisionSynonyms(t *testing.T) { + store := seedWaitingExecution(t, "exec-1", "agent-1", "req-abc") + router := resolveApprovalRouter(store) + + resp := postResolveApproval(t, router, "exec-1", map[string]any{"decision": "approve"}) + + require.Equal(t, http.StatusOK, resp.Code) + + var result map[string]any + require.NoError(t, json.Unmarshal(resp.Body.Bytes(), &result)) + assert.Equal(t, "approved", result["decision"]) +} + +// Contract: an unknown decision is a 400 with a structured error code. +func TestResolveApproval_InvalidDecision(t *testing.T) { + store := seedWaitingExecution(t, "exec-1", "agent-1", "req-abc") + router := resolveApprovalRouter(store) + + resp := postResolveApproval(t, router, "exec-1", map[string]any{"decision": "maybe"}) + + require.Equal(t, http.StatusBadRequest, resp.Code) + + var result map[string]any + require.NoError(t, json.Unmarshal(resp.Body.Bytes(), &result)) + assert.Equal(t, "invalid_decision", result["error"]) +} + +// Contract: "expired" is not an operator decision on this endpoint. +func TestResolveApproval_ExpiredRejected(t *testing.T) { + store := seedWaitingExecution(t, "exec-1", "agent-1", "req-abc") + router := resolveApprovalRouter(store) + + resp := postResolveApproval(t, router, "exec-1", map[string]any{"decision": "expired"}) + + require.Equal(t, http.StatusBadRequest, resp.Code) +} + +// Contract: missing decision fails request binding with a 400. +func TestResolveApproval_MissingDecision(t *testing.T) { + store := seedWaitingExecution(t, "exec-1", "agent-1", "req-abc") + router := resolveApprovalRouter(store) + + resp := postResolveApproval(t, router, "exec-1", map[string]any{}) + + require.Equal(t, http.StatusBadRequest, resp.Code) +} + +// Contract: unknown execution ID is a 404, not a panic or 500. +func TestResolveApproval_ExecutionNotFound(t *testing.T) { + store := seedWaitingExecution(t, "exec-1", "agent-1", "req-abc") + router := resolveApprovalRouter(store) + + resp := postResolveApproval(t, router, "exec-missing", map[string]any{"decision": "approved"}) + + require.Equal(t, http.StatusNotFound, resp.Code) +} + +// Contract: an execution without a pending approval request is a 404 with the +// same no_approval_request code approval-status uses. +func TestResolveApproval_NoApprovalRequest(t *testing.T) { + agent := &types.AgentNode{ID: "agent-1"} + store := newTestExecutionStorage(agent) + + now := time.Now().UTC() + runID := "run-1" + require.NoError(t, store.CreateExecutionRecord(context.Background(), &types.Execution{ + ExecutionID: "exec-1", + RunID: runID, + AgentNodeID: "agent-1", + Status: types.ExecutionStatusRunning, + StartedAt: now, + CreatedAt: now, + })) + require.NoError(t, store.StoreWorkflowExecution(context.Background(), &types.WorkflowExecution{ + ExecutionID: "exec-1", + WorkflowID: "wf-1", + RunID: &runID, + AgentNodeID: "agent-1", + Status: types.ExecutionStatusRunning, + StartedAt: now, + })) + + router := resolveApprovalRouter(store) + resp := postResolveApproval(t, router, "exec-1", map[string]any{"decision": "approved"}) + + require.Equal(t, http.StatusNotFound, resp.Code) + + var result map[string]any + require.NoError(t, json.Unmarshal(resp.Body.Bytes(), &result)) + assert.Equal(t, "no_approval_request", result["error"]) +} + +// Contract: resolving twice is idempotent — the second call reports +// already_processed and does not change state again. +func TestResolveApproval_Idempotent(t *testing.T) { + store := seedWaitingExecution(t, "exec-1", "agent-1", "req-abc") + router := resolveApprovalRouter(store) + + first := postResolveApproval(t, router, "exec-1", map[string]any{"decision": "approved"}) + require.Equal(t, http.StatusOK, first.Code) + + second := postResolveApproval(t, router, "exec-1", map[string]any{"decision": "rejected"}) + require.Equal(t, http.StatusOK, second.Code) + + var result map[string]any + require.NoError(t, json.Unmarshal(second.Body.Bytes(), &result)) + assert.Equal(t, "already_processed", result["status"]) + + wfExec, err := store.GetWorkflowExecution(context.Background(), "exec-1") + require.NoError(t, err) + assert.Equal(t, types.ExecutionStatusRunning, wfExec.Status) + require.NotNil(t, wfExec.ApprovalStatus) + assert.Equal(t, "approved", *wfExec.ApprovalStatus) +} diff --git a/control-plane/internal/handlers/webhook_approval.go b/control-plane/internal/handlers/webhook_approval.go index 8efac6aec..8e9d23fb0 100644 --- a/control-plane/internal/handlers/webhook_approval.go +++ b/control-plane/internal/handlers/webhook_approval.go @@ -192,8 +192,6 @@ func (c *webhookApprovalController) handleApprovalWebhook(ctx *gin.Context) { return } - reqCtx := ctx.Request.Context() - // Find the workflow execution by approval_request_id executionID, wfExec, err := c.findExecutionByApprovalRequestID(ctx, payload.RequestID) if err != nil { @@ -207,6 +205,19 @@ func (c *webhookApprovalController) handleApprovalWebhook(ctx *gin.Context) { return } + c.applyApprovalDecision(ctx, executionID, wfExec, payload, decision) +} + +// applyApprovalDecision applies a validated approval decision to a waiting +// execution: idempotency check, state transitions, approval bookkeeping, +// observability events, and the async agent callback. Shared by the +// HMAC-signed webhook and the authenticated +// POST /executions/:execution_id/approval-response endpoint so both paths +// resolve approvals identically. decision must already be normalized to one +// of approved, rejected, request_changes, or expired. +func (c *webhookApprovalController) applyApprovalDecision(ctx *gin.Context, executionID string, wfExec *types.WorkflowExecution, payload *ApprovalWebhookPayload, decision string) { + reqCtx := ctx.Request.Context() + // Idempotency: if execution is no longer in waiting state, it was already // processed by a previous webhook delivery. Return 200 so the sender // (hax-sdk retry queue) considers it delivered and stops retrying. @@ -303,7 +314,7 @@ func (c *webhookApprovalController) handleApprovalWebhook(ctx *gin.Context) { } // Update the workflow execution with approval resolution (authoritative — must not lose the decision) - err = c.store.UpdateWorkflowExecution(reqCtx, executionID, func(current *types.WorkflowExecution) (*types.WorkflowExecution, error) { + err := c.store.UpdateWorkflowExecution(reqCtx, executionID, func(current *types.WorkflowExecution) (*types.WorkflowExecution, error) { if current == nil { return nil, fmt.Errorf("execution %s not found", executionID) } diff --git a/control-plane/internal/server/routes_core.go b/control-plane/internal/server/routes_core.go index 1e9988382..7cc23ab3e 100644 --- a/control-plane/internal/server/routes_core.go +++ b/control-plane/internal/server/routes_core.go @@ -131,6 +131,10 @@ func (s *AgentFieldServer) registerCoreRoutes(agentAPI *gin.RouterGroup) { // agents handle external approval service communication directly. agentAPI.POST("/executions/:execution_id/request-approval", handlers.RequestApprovalHandler(s.storage)) agentAPI.GET("/executions/:execution_id/approval-status", handlers.GetApprovalStatusHandler(s.storage)) + // Authenticated approval resolution by execution ID — same resolution + // logic as the HMAC webhook below, for operators/CLIs that hold an API + // key rather than the webhook secret. + agentAPI.POST("/executions/:execution_id/approval-response", handlers.ResolveApprovalHandler(s.storage)) // Agent-scoped approval routes — enforce that the execution belongs to the requesting agent. agentAPI.POST("/agents/:node_id/executions/:execution_id/request-approval", handlers.AgentScopedRequestApprovalHandler(s.storage)) From 51c898df55c477791aeb9e3e288d9c40eda3d573 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Fri, 10 Jul 2026 15:30:09 -0400 Subject: [PATCH 2/9] feat(control-plane): af agent exec steering verbs in agent-mode CLI MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add 'af agent exec pause|resume|cancel|restart|approval-status|approve' as thin wrappers over the existing /api/v1/executions/:execution_id/* endpoints, emitting the standard AgentResponse envelope with non-zero exit on error. approve wires --decision approved|rejected| request_changes [--reason] to the new authenticated approval-response endpoint. proxyToServer now normalizes legacy string errors ({"error":"..."}, optionally with a sibling message) into the structured envelope error object {code,message,hint}, deriving the code from the HTTP status when the handler didn't provide one — so agents scripting against 'af agent' always get {ok:false,error:{...}} on failures. Co-Authored-By: Claude Fable 5 --- control-plane/internal/cli/agent_commands.go | 103 +++++++- control-plane/internal/cli/agent_exec.go | 184 +++++++++++++++ control-plane/internal/cli/agent_exec_test.go | 222 ++++++++++++++++++ 3 files changed, 508 insertions(+), 1 deletion(-) create mode 100644 control-plane/internal/cli/agent_exec.go create mode 100644 control-plane/internal/cli/agent_exec_test.go diff --git a/control-plane/internal/cli/agent_commands.go b/control-plane/internal/cli/agent_commands.go index af5bdaf22..f8fe02e00 100644 --- a/control-plane/internal/cli/agent_commands.go +++ b/control-plane/internal/cli/agent_commands.go @@ -69,6 +69,7 @@ func NewAgentCommand() *cobra.Command { cmd.AddCommand(newAgentDiscoverCmd()) cmd.AddCommand(newAgentQueryCmd()) cmd.AddCommand(newAgentRunCmd()) + cmd.AddCommand(newAgentExecCmd()) cmd.AddCommand(newAgentAgentSummaryCmd()) cmd.AddCommand(newAgentKBCmd()) cmd.AddCommand(newAgentBatchCmd()) @@ -550,6 +551,8 @@ func proxyToServer(method, path string, body interface{}) { if _, hasHint := errMap["hint"]; !hasHint { errMap["hint"] = defaultHintForStatus(statusCode) } + } else if errStr, ok := payload["error"].(string); ok { + payload["error"] = structuredErrorFromString(payload, errStr, statusCode) } } @@ -596,6 +599,44 @@ func readBatchInput(file string) ([]byte, error) { return data, nil } +// structuredErrorFromString converts legacy {"error":"..."} responses into the +// structured envelope error object {code,message,hint}. Handlers that return +// {"error":"some_code","message":"detail"} keep the error string as the code; +// otherwise the code is derived from the HTTP status. +func structuredErrorFromString(payload map[string]interface{}, errStr string, statusCode int) map[string]interface{} { + code := defaultCodeForStatus(statusCode) + message := errStr + if msg, ok := payload["message"].(string); ok && strings.TrimSpace(msg) != "" { + code = errStr + message = msg + } + return map[string]interface{}{ + "code": code, + "message": message, + "hint": defaultHintForStatus(statusCode), + } +} + +func defaultCodeForStatus(statusCode int) string { + switch statusCode { + case http.StatusBadRequest: + return "bad_request" + case http.StatusUnauthorized: + return "unauthorized" + case http.StatusForbidden: + return "forbidden" + case http.StatusNotFound: + return "not_found" + case http.StatusConflict: + return "conflict" + default: + if statusCode >= 500 { + return "server_error" + } + return "request_failed" + } +} + func defaultHintForStatus(statusCode int) string { switch statusCode { case http.StatusUnauthorized: @@ -689,6 +730,66 @@ func agentHelpData() map[string]interface{} { "flags": []map[string]string{{"name": "id", "short": "", "type": "string (required)"}}, "example": "af agent agent-summary --id sec-analyst", }, + { + "name": "exec pause", + "description": "Pause a running execution", + "usage": "af agent exec pause --id [--reason ]", + "flags": []map[string]string{ + {"name": "id", "short": "", "type": "string (required)"}, + {"name": "reason", "short": "", "type": "string"}, + }, + "example": "af agent exec pause --id exec_123 --reason 'manual review'", + }, + { + "name": "exec resume", + "description": "Resume a paused execution", + "usage": "af agent exec resume --id ", + "flags": []map[string]string{{"name": "id", "short": "", "type": "string (required)"}}, + "example": "af agent exec resume --id exec_123", + }, + { + "name": "exec cancel", + "description": "Cancel an execution", + "usage": "af agent exec cancel --id [--reason ]", + "flags": []map[string]string{ + {"name": "id", "short": "", "type": "string (required)"}, + {"name": "reason", "short": "", "type": "string"}, + }, + "example": "af agent exec cancel --id exec_123 --reason 'wrong input'", + }, + { + "name": "exec restart", + "description": "Restart a workflow from an execution point", + "usage": "af agent exec restart --id [--scope] [--reuse] [--fork] [--model] [--input] [--reason]", + "flags": []map[string]string{ + {"name": "id", "short": "", "type": "string (required)"}, + {"name": "scope", "short": "", "type": "string (workflow|execution)"}, + {"name": "reuse", "short": "", "type": "string (succeeded-before|all-succeeded|none)"}, + {"name": "fork", "short": "", "type": "bool"}, + {"name": "model", "short": "", "type": "string"}, + {"name": "input", "short": "", "type": "string (JSON or @path)"}, + {"name": "reason", "short": "", "type": "string"}, + }, + "example": "af agent exec restart --id exec_123 --scope workflow --reuse succeeded-before", + }, + { + "name": "exec approval-status", + "description": "Get the approval status for an execution", + "usage": "af agent exec approval-status --id ", + "flags": []map[string]string{{"name": "id", "short": "", "type": "string (required)"}}, + "example": "af agent exec approval-status --id exec_123", + }, + { + "name": "exec approve", + "description": "Resolve a pending approval for an execution", + "usage": "af agent exec approve --id --decision [--reason ]", + "flags": []map[string]string{ + {"name": "id", "short": "", "type": "string (required)"}, + {"name": "decision", "short": "", "type": "string (required)"}, + {"name": "reason", "short": "", "type": "string"}, + }, + "example": "af agent exec approve --id exec_123 --decision approved", + }, { "name": "kb topics", "description": "List knowledge base topics", @@ -749,7 +850,7 @@ func agentHelpData() map[string]interface{} { "auth": map[string]interface{}{ "method": "Set X-API-Key header via --api-key or AGENTFIELD_API_KEY", "public_endpoints": []string{"GET /api/v1/agentic/kb/topics", "GET /api/v1/agentic/kb/articles", "GET /api/v1/agentic/kb/articles/:article_id", "GET /api/v1/agentic/kb/guide"}, - "requires_auth": []string{"GET /api/v1/agentic/status", "GET /api/v1/agentic/discover", "POST /api/v1/agentic/query", "GET /api/v1/agentic/run/:run_id", "GET /api/v1/agentic/agent/:agent_id/summary", "POST /api/v1/agentic/batch"}, + "requires_auth": []string{"GET /api/v1/agentic/status", "GET /api/v1/agentic/discover", "POST /api/v1/agentic/query", "GET /api/v1/agentic/run/:run_id", "GET /api/v1/agentic/agent/:agent_id/summary", "POST /api/v1/agentic/batch", "POST /api/v1/executions/:execution_id/pause", "POST /api/v1/executions/:execution_id/resume", "POST /api/v1/executions/:execution_id/cancel", "POST /api/v1/executions/:execution_id/restart", "GET /api/v1/executions/:execution_id/approval-status", "POST /api/v1/executions/:execution_id/approval-response"}, }, "response_schemas": map[string]interface{}{ "success": map[string]interface{}{ diff --git a/control-plane/internal/cli/agent_exec.go b/control-plane/internal/cli/agent_exec.go new file mode 100644 index 000000000..457bd4070 --- /dev/null +++ b/control-plane/internal/cli/agent_exec.go @@ -0,0 +1,184 @@ +package cli + +import ( + "fmt" + "net/http" + "net/url" + "strings" + + "github.com/spf13/cobra" +) + +// newAgentExecCmd groups execution steering verbs for agent mode. Each verb is +// a thin wrapper over the existing /api/v1/executions/:execution_id/* REST +// endpoints and emits the standard AgentResponse JSON envelope. +func newAgentExecCmd() *cobra.Command { + execCmd := &cobra.Command{ + Use: "exec", + Short: "Steer executions: pause, resume, cancel, restart, approvals", + Run: func(cmd *cobra.Command, args []string) { + agentOutput(map[string]interface{}{ + "message": "Use an exec subcommand", + "available": []string{ + "pause", + "resume", + "cancel", + "restart", + "approval-status", + "approve", + }, + }) + }, + } + + execCmd.AddCommand(newAgentExecActionCmd("pause", "Pause a running execution", true)) + execCmd.AddCommand(newAgentExecActionCmd("resume", "Resume a paused execution", false)) + execCmd.AddCommand(newAgentExecActionCmd("cancel", "Cancel an execution", true)) + execCmd.AddCommand(newAgentExecRestartCmd()) + execCmd.AddCommand(newAgentExecApprovalStatusCmd()) + execCmd.AddCommand(newAgentExecApproveCmd()) + + return execCmd +} + +// requireExecutionID trims and validates the --id flag shared by exec verbs. +// On failure it emits an error envelope and exits non-zero via agentError. +func requireExecutionID(executionID, verb string) string { + trimmed := strings.TrimSpace(executionID) + if trimmed == "" { + agentError( + "missing_required_flag", + "--id is required", + fmt.Sprintf("Provide an execution ID, for example: af agent exec %s --id exec_123", verb), + ) + } + return trimmed +} + +// newAgentExecActionCmd builds pause/resume/cancel verbs. withReason controls +// whether an optional --reason flag is sent in the request body. +func newAgentExecActionCmd(action, short string, withReason bool) *cobra.Command { + var executionID string + var reason string + + cmd := &cobra.Command{ + Use: action, + Short: short, + Run: func(cmd *cobra.Command, args []string) { + id := requireExecutionID(executionID, action) + + var body interface{} + if withReason { + payload := map[string]string{} + if v := strings.TrimSpace(reason); v != "" { + payload["reason"] = v + } + body = payload + } + + proxyToServer(http.MethodPost, "/api/v1/executions/"+url.PathEscape(id)+"/"+action, body) + }, + } + + cmd.Flags().StringVar(&executionID, "id", "", "Execution ID") + if withReason { + cmd.Flags().StringVar(&reason, "reason", "", "Reason for the action") + } + return cmd +} + +func newAgentExecRestartCmd() *cobra.Command { + var executionID string + opts := executionActionOptions{ + scope: "workflow", + reuse: "succeeded-before", + } + + cmd := &cobra.Command{ + Use: "restart", + Short: "Restart a workflow from an execution point", + Run: func(cmd *cobra.Command, args []string) { + id := requireExecutionID(executionID, "restart") + + body, err := buildRestartExecutionBody(opts) + if err != nil { + agentError( + "invalid_flag_value", + err.Error(), + "Pass valid JSON to --input (inline or @path) and check the restart flags.", + ) + } + + proxyToServer(http.MethodPost, "/api/v1/executions/"+url.PathEscape(id)+"/restart", body) + }, + } + + cmd.Flags().StringVar(&executionID, "id", "", "Execution ID") + cmd.Flags().StringVar(&opts.scope, "scope", opts.scope, "Restart scope: workflow or execution") + cmd.Flags().StringVar(&opts.reuse, "reuse", opts.reuse, "Replay reuse mode: succeeded-before, all-succeeded, or none") + cmd.Flags().BoolVar(&opts.fork, "fork", false, "Mark this restart as a fork with intentional changes") + cmd.Flags().StringVar(&opts.model, "model", "", "Model override to send in restart context") + cmd.Flags().StringVar(&opts.input, "input", "", "JSON input override or @path to a JSON file") + cmd.Flags().StringVar(&opts.reason, "reason", "", "Reason for restarting the execution") + return cmd +} + +func newAgentExecApprovalStatusCmd() *cobra.Command { + var executionID string + + cmd := &cobra.Command{ + Use: "approval-status", + Short: "Get the approval status for an execution", + Run: func(cmd *cobra.Command, args []string) { + id := requireExecutionID(executionID, "approval-status") + proxyToServer(http.MethodGet, "/api/v1/executions/"+url.PathEscape(id)+"/approval-status", nil) + }, + } + + cmd.Flags().StringVar(&executionID, "id", "", "Execution ID") + return cmd +} + +func newAgentExecApproveCmd() *cobra.Command { + var executionID string + var decision string + var reason string + + cmd := &cobra.Command{ + Use: "approve", + Short: "Resolve a pending approval for an execution", + Run: func(cmd *cobra.Command, args []string) { + id := requireExecutionID(executionID, "approve --decision approved") + + decision = strings.TrimSpace(decision) + if decision == "" { + agentError( + "missing_required_flag", + "--decision is required", + "Set --decision to approved, rejected, or request_changes.", + ) + } + switch decision { + case "approved", "rejected", "request_changes": + default: + agentError( + "invalid_flag_value", + fmt.Sprintf("invalid --decision %q", decision), + "Set --decision to approved, rejected, or request_changes.", + ) + } + + payload := map[string]string{"decision": decision} + if v := strings.TrimSpace(reason); v != "" { + payload["reason"] = v + } + + proxyToServer(http.MethodPost, "/api/v1/executions/"+url.PathEscape(id)+"/approval-response", payload) + }, + } + + cmd.Flags().StringVar(&executionID, "id", "", "Execution ID") + cmd.Flags().StringVar(&decision, "decision", "", "Approval decision: approved, rejected, or request_changes") + cmd.Flags().StringVar(&reason, "reason", "", "Optional feedback recorded with the decision") + return cmd +} diff --git a/control-plane/internal/cli/agent_exec_test.go b/control-plane/internal/cli/agent_exec_test.go new file mode 100644 index 000000000..1bc3506b0 --- /dev/null +++ b/control-plane/internal/cli/agent_exec_test.go @@ -0,0 +1,222 @@ +package cli + +import ( + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "os" + "os/exec" + "strings" + "testing" + + "github.com/stretchr/testify/require" +) + +// Contract: every `af agent exec --id X` call proxies to the matching +// /api/v1/executions/X/ endpoint and emits an AgentResponse envelope +// with ok:true and meta.status_code on 2xx. +func TestAgentExecHappyPaths(t *testing.T) { + tests := []struct { + name string + args []string + wantMethod string + wantPath string + wantBody map[string]interface{} + }{ + { + name: "pause with reason", + args: []string{"exec", "pause", "--id", "exec_1", "--reason", "manual review"}, + wantMethod: http.MethodPost, + wantPath: "/api/v1/executions/exec_1/pause", + wantBody: map[string]interface{}{"reason": "manual review"}, + }, + { + name: "pause without reason sends empty object", + args: []string{"exec", "pause", "--id", "exec_1"}, + wantMethod: http.MethodPost, + wantPath: "/api/v1/executions/exec_1/pause", + wantBody: map[string]interface{}{}, + }, + { + name: "resume has no body", + args: []string{"exec", "resume", "--id", "exec_1"}, + wantMethod: http.MethodPost, + wantPath: "/api/v1/executions/exec_1/resume", + wantBody: nil, + }, + { + name: "cancel with reason", + args: []string{"exec", "cancel", "--id", "exec_1", "--reason", "wrong input"}, + wantMethod: http.MethodPost, + wantPath: "/api/v1/executions/exec_1/cancel", + wantBody: map[string]interface{}{"reason": "wrong input"}, + }, + { + name: "restart sends default scope and reuse", + args: []string{"exec", "restart", "--id", "exec_1"}, + wantMethod: http.MethodPost, + wantPath: "/api/v1/executions/exec_1/restart", + wantBody: map[string]interface{}{"scope": "workflow", "reuse": "succeeded-before"}, + }, + { + name: "approval status is a GET", + args: []string{"exec", "approval-status", "--id", "exec_1"}, + wantMethod: http.MethodGet, + wantPath: "/api/v1/executions/exec_1/approval-status", + wantBody: nil, + }, + { + name: "approve sends decision and reason", + args: []string{"exec", "approve", "--id", "exec_1", "--decision", "approved", "--reason", "lgtm"}, + wantMethod: http.MethodPost, + wantPath: "/api/v1/executions/exec_1/approval-response", + wantBody: map[string]interface{}{"decision": "approved", "reason": "lgtm"}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var gotMethod, gotPath string + var gotBody []byte + + oldTransport := http.DefaultTransport + http.DefaultTransport = roundTripFunc(func(req *http.Request) (*http.Response, error) { + gotMethod = req.Method + gotPath = req.URL.Path + if req.Body != nil { + gotBody, _ = io.ReadAll(req.Body) + } + return &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"application/json"}}, + Body: io.NopCloser(strings.NewReader(`{"execution_id":"exec_1","status":"ok"}`)), + Request: req, + }, nil + }) + defer func() { http.DefaultTransport = oldTransport }() + + oldServer, oldFormat, oldTimeout := serverURL, outputFormat, requestTimeout + serverURL, outputFormat, requestTimeout = "http://agent.test", "json", 1 + defer func() { + serverURL, outputFormat, requestTimeout = oldServer, oldFormat, oldTimeout + }() + + output := captureOutput(t, func() { + cmd := NewAgentCommand() + cmd.SetArgs(tt.args) + require.NoError(t, cmd.Execute()) + }) + + require.Equal(t, tt.wantMethod, gotMethod) + require.Equal(t, tt.wantPath, gotPath) + + if tt.wantBody == nil { + require.Empty(t, gotBody) + } else { + var decoded map[string]interface{} + require.NoError(t, json.Unmarshal(gotBody, &decoded)) + require.Equal(t, tt.wantBody, decoded) + } + + require.Contains(t, output, `"ok": true`) + require.Contains(t, output, `"status_code": 200`) + require.Contains(t, output, `"execution_id": "exec_1"`) + }) + } +} + +// Contract: exec subcommand without a verb lists the available verbs. +func TestAgentExecListsVerbs(t *testing.T) { + oldFormat := outputFormat + outputFormat = "json" + defer func() { outputFormat = oldFormat }() + + output := captureOutput(t, func() { + cmd := NewAgentCommand() + cmd.SetArgs([]string{"exec"}) + require.NoError(t, cmd.Execute()) + }) + for _, verb := range []string{"pause", "resume", "cancel", "restart", "approval-status", "approve"} { + require.Contains(t, output, verb) + } +} + +// Contract: validation and server error paths exit non-zero and print a +// structured {ok:false,error:{code,message,hint}} envelope. +func TestAgentExecExitOutputs(t *testing.T) { + cases := []struct { + name string + wants []string + }{ + {name: "exec-pause-missing-id", wants: []string{`"code": "missing_required_flag"`, `"--id is required"`}}, + {name: "exec-approve-missing-decision", wants: []string{`"code": "missing_required_flag"`, `"--decision is required"`}}, + {name: "exec-approve-invalid-decision", wants: []string{`"code": "invalid_flag_value"`}}, + {name: "exec-restart-invalid-input", wants: []string{`"code": "invalid_flag_value"`}}, + {name: "exec-pause-not-found", wants: []string{`"ok": false`, `"code": "not_found"`, `"status_code": 404`, "execution exec_x not found"}}, + {name: "exec-resume-conflict-with-code", wants: []string{`"ok": false`, `"code": "invalid_state"`, `"status_code": 409`}}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + out, err := runAgentExecExitHelper(t, tc.name) + require.Error(t, err) + exitErr := &exec.ExitError{} + require.ErrorAs(t, err, &exitErr) + for _, want := range tc.wants { + require.Contains(t, out, want) + } + }) + } +} + +func runAgentExecExitHelper(t *testing.T, mode string) (string, error) { + t.Helper() + + cmd := exec.Command(os.Args[0], "-test.run=TestAgentExecExitHelper", "--", mode) + cmd.Env = append(os.Environ(), "GO_WANT_AGENT_EXEC_EXIT_HELPER=1") + out, err := cmd.CombinedOutput() + return string(out), err +} + +func TestAgentExecExitHelper(t *testing.T) { + if os.Getenv("GO_WANT_AGENT_EXEC_EXIT_HELPER") != "1" { + return + } + + runAgent := func(args ...string) { + cmd := NewAgentCommand() + cmd.SetArgs(args) + _ = cmd.Execute() + } + + mode := os.Args[len(os.Args)-1] + switch mode { + case "exec-pause-missing-id": + runAgent("exec", "pause") + case "exec-approve-missing-decision": + runAgent("exec", "approve", "--id", "exec_x") + case "exec-approve-invalid-decision": + runAgent("exec", "approve", "--id", "exec_x", "--decision", "maybe") + case "exec-restart-invalid-input": + runAgent("exec", "restart", "--id", "exec_x", "--input", "{not-json") + case "exec-pause-not-found": + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusNotFound) + _, _ = w.Write([]byte(`{"error":"execution exec_x not found"}`)) + })) + defer server.Close() + serverURL, requestTimeout = server.URL, 1 + runAgent("exec", "pause", "--id", "exec_x") + case "exec-resume-conflict-with-code": + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusConflict) + _, _ = w.Write([]byte(`{"error":"invalid_state","message":"execution is in 'running' state; must be 'paused'"}`)) + })) + defer server.Close() + serverURL, requestTimeout = server.URL, 1 + runAgent("exec", "resume", "--id", "exec_x") + } +} From 2bfc327947e6ee5609af2c810c12844fc1201e28 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Fri, 10 Jul 2026 15:30:33 -0400 Subject: [PATCH 3/9] feat(control-plane): events resource in agentic unified query MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Expose persisted execution lifecycle events (workflow_execution_events) through POST /api/v1/agentic/query as resource=events so agents can poll event history from a shell loop instead of consuming SSE. Queries require filters.execution_id (direct per-execution listing) or filters.run_id (fan-out over the run's execution records); results are sorted by emitted_at ascending with sequence tie-breaking, honor since/until RFC3339 bounds, and paginate with limit/offset (total is the pre-pagination count). No new persistence layer or storage interface methods — this composes the existing QueryExecutionRecords and ListWorkflowExecutionEvents queries. CLI: 'af agent query -r events --execution-id X | --run-id Y' with a new --execution-id filter flag; help documents that this is a pollable snapshot of the SSE stream, not a live subscription. Co-Authored-By: Claude Fable 5 --- control-plane/internal/cli/agent_commands.go | 24 +- control-plane/internal/cli/agent_exec_test.go | 39 +++ .../internal/handlers/agentic/query.go | 113 +++++++- .../handlers/agentic/query_events_test.go | 256 ++++++++++++++++++ 4 files changed, 419 insertions(+), 13 deletions(-) create mode 100644 control-plane/internal/handlers/agentic/query_events_test.go diff --git a/control-plane/internal/cli/agent_commands.go b/control-plane/internal/cli/agent_commands.go index f8fe02e00..0b080c77c 100644 --- a/control-plane/internal/cli/agent_commands.go +++ b/control-plane/internal/cli/agent_commands.go @@ -133,6 +133,7 @@ func newAgentQueryCmd() *cobra.Command { var status string var agentID string var runID string + var executionID string var since string var until string var limit int @@ -142,10 +143,19 @@ func newAgentQueryCmd() *cobra.Command { cmd := &cobra.Command{ Use: "query", Short: "Run unified resource query", + Long: `Run a unified resource query against the control plane. + +Resources: runs, executions, agents, workflows, sessions, events. + +The events resource returns persisted execution lifecycle events +(status transitions, approval waits/resolutions) sorted by emitted_at +ascending — a pollable snapshot of the SSE event stream, filtered by +--execution-id or fanned out across a run with --run-id (one of the two +is required). It does not include live in-flight SSE-only data.`, Run: func(cmd *cobra.Command, args []string) { resource = strings.TrimSpace(resource) if resource == "" { - agentError("missing_required_flag", "--resource is required", "Set --resource to one of runs, executions, agents, workflows, sessions.") + agentError("missing_required_flag", "--resource is required", "Set --resource to one of runs, executions, agents, workflows, sessions, events.") } filters := map[string]string{} @@ -158,6 +168,9 @@ func newAgentQueryCmd() *cobra.Command { if v := strings.TrimSpace(runID); v != "" { filters["run_id"] = v } + if v := strings.TrimSpace(executionID); v != "" { + filters["execution_id"] = v + } if v := strings.TrimSpace(since); v != "" { filters["since"] = v } @@ -189,10 +202,11 @@ func newAgentQueryCmd() *cobra.Command { }, } - cmd.Flags().StringVarP(&resource, "resource", "r", "", "Resource type: runs, executions, agents, workflows, sessions") + cmd.Flags().StringVarP(&resource, "resource", "r", "", "Resource type: runs, executions, agents, workflows, sessions, events") cmd.Flags().StringVar(&status, "status", "", "Status filter") cmd.Flags().StringVar(&agentID, "agent-id", "", "Agent ID filter") cmd.Flags().StringVar(&runID, "run-id", "", "Run ID filter") + cmd.Flags().StringVar(&executionID, "execution-id", "", "Execution ID filter (events resource)") cmd.Flags().StringVar(&since, "since", "", "RFC3339 lower time bound") cmd.Flags().StringVar(&until, "until", "", "RFC3339 upper time bound") cmd.Flags().IntVar(&limit, "limit", 20, "Result limit") @@ -701,13 +715,14 @@ func agentHelpData() map[string]interface{} { }, { "name": "query", - "description": "Query runs/executions/agents/workflows/sessions", - "usage": "af agent query --resource runs [--status] [--agent-id] [--run-id] [--since] [--until] [--limit] [--offset] [--include]", + "description": "Query runs/executions/agents/workflows/sessions/events", + "usage": "af agent query --resource runs [--status] [--agent-id] [--run-id] [--execution-id] [--since] [--until] [--limit] [--offset] [--include]", "flags": []map[string]string{ {"name": "resource", "short": "r", "type": "string (required)"}, {"name": "status", "short": "", "type": "string"}, {"name": "agent-id", "short": "", "type": "string"}, {"name": "run-id", "short": "", "type": "string"}, + {"name": "execution-id", "short": "", "type": "string"}, {"name": "since", "short": "", "type": "string(RFC3339)"}, {"name": "until", "short": "", "type": "string(RFC3339)"}, {"name": "limit", "short": "", "type": "int"}, @@ -715,6 +730,7 @@ func agentHelpData() map[string]interface{} { {"name": "include", "short": "", "type": "string(csv)"}, }, "example": "af agent query -r executions --agent-id analyst --status completed --limit 25", + "notes": "resource=events returns persisted execution lifecycle events sorted by emitted_at ascending; requires --execution-id or --run-id. It is a pollable snapshot of the SSE stream, not a live subscription.", }, { "name": "run", diff --git a/control-plane/internal/cli/agent_exec_test.go b/control-plane/internal/cli/agent_exec_test.go index 1bc3506b0..0988aa704 100644 --- a/control-plane/internal/cli/agent_exec_test.go +++ b/control-plane/internal/cli/agent_exec_test.go @@ -126,6 +126,45 @@ func TestAgentExecHappyPaths(t *testing.T) { } } +// Contract: `af agent query -r events` forwards execution_id/run_id/since/ +// until filters to POST /api/v1/agentic/query. +func TestAgentQueryEventsSendsFilters(t *testing.T) { + var gotBody []byte + + oldTransport := http.DefaultTransport + http.DefaultTransport = roundTripFunc(func(req *http.Request) (*http.Response, error) { + gotBody, _ = io.ReadAll(req.Body) + return &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"application/json"}}, + Body: io.NopCloser(strings.NewReader(`{"ok":true,"data":{"resource":"events","results":[]}}`)), + Request: req, + }, nil + }) + defer func() { http.DefaultTransport = oldTransport }() + + oldServer, oldFormat, oldTimeout := serverURL, outputFormat, requestTimeout + serverURL, outputFormat, requestTimeout = "http://agent.test", "json", 1 + defer func() { + serverURL, outputFormat, requestTimeout = oldServer, oldFormat, oldTimeout + }() + + _ = captureOutput(t, func() { + cmd := NewAgentCommand() + cmd.SetArgs([]string{"query", "-r", "events", "--execution-id", "exec_1", "--run-id", "run_1", "--since", "2026-07-01T00:00:00Z", "--limit", "5"}) + require.NoError(t, cmd.Execute()) + }) + + var payload map[string]interface{} + require.NoError(t, json.Unmarshal(gotBody, &payload)) + require.Equal(t, "events", payload["resource"]) + filters := payload["filters"].(map[string]interface{}) + require.Equal(t, "exec_1", filters["execution_id"]) + require.Equal(t, "run_1", filters["run_id"]) + require.Equal(t, "2026-07-01T00:00:00Z", filters["since"]) + require.Equal(t, float64(5), payload["limit"]) +} + // Contract: exec subcommand without a verb lists the available verbs. func TestAgentExecListsVerbs(t *testing.T) { oldFormat := outputFormat diff --git a/control-plane/internal/handlers/agentic/query.go b/control-plane/internal/handlers/agentic/query.go index 951bef371..ac738ecbb 100644 --- a/control-plane/internal/handlers/agentic/query.go +++ b/control-plane/internal/handlers/agentic/query.go @@ -3,6 +3,8 @@ package agentic import ( "context" "net/http" + "sort" + "strings" "time" "github.com/Agent-Field/agentfield/control-plane/internal/storage" @@ -12,7 +14,7 @@ import ( // QueryRequest is the body for POST /agentic/query. type QueryRequest struct { - Resource string `json:"resource" binding:"required"` // "runs", "executions", "agents", "workflows", "sessions" + Resource string `json:"resource" binding:"required"` // "runs", "executions", "agents", "workflows", "sessions", "events" Filters QueryFilters `json:"filters"` Include []string `json:"include,omitempty"` // related resources to include Limit int `json:"limit"` @@ -21,13 +23,14 @@ type QueryRequest struct { // QueryFilters contains optional filter criteria. type QueryFilters struct { - Status string `json:"status,omitempty"` - AgentID string `json:"agent_id,omitempty"` - RunID string `json:"run_id,omitempty"` - SessionID string `json:"session_id,omitempty"` - ActorID string `json:"actor_id,omitempty"` - Since string `json:"since,omitempty"` // RFC3339 - Until string `json:"until,omitempty"` // RFC3339 + Status string `json:"status,omitempty"` + AgentID string `json:"agent_id,omitempty"` + RunID string `json:"run_id,omitempty"` + ExecutionID string `json:"execution_id,omitempty"` + SessionID string `json:"session_id,omitempty"` + ActorID string `json:"actor_id,omitempty"` + Since string `json:"since,omitempty"` // RFC3339 + Until string `json:"until,omitempty"` // RFC3339 } // QueryHandler handles unified resource queries. @@ -56,9 +59,11 @@ func QueryHandler(store storage.StorageProvider) gin.HandlerFunc { queryWorkflows(c, ctx, store, req) case "sessions": querySessions(c, ctx, store, req) + case "events": + queryEvents(c, ctx, store, req) default: respondError(c, http.StatusBadRequest, "invalid_resource", - "resource must be one of: runs, executions, agents, workflows, sessions") + "resource must be one of: runs, executions, agents, workflows, sessions, events") } } } @@ -216,6 +221,96 @@ func queryWorkflows(c *gin.Context, ctx context.Context, store storage.StoragePr }) } +// queryEvents exposes persisted execution lifecycle events +// (workflow_execution_events) as a time-sorted, pollable list. It composes the +// existing per-execution storage query: an execution_id filter reads that +// execution's events directly; a run_id filter fans out over the run's +// execution records. Results are sorted by emitted_at ascending (ties broken +// by sequence). This surfaces coarse lifecycle transitions — it is a polling +// snapshot, not the live SSE stream at /executions/:execution_id/events. +func queryEvents(c *gin.Context, ctx context.Context, store storage.StorageProvider, req QueryRequest) { + executionID := strings.TrimSpace(req.Filters.ExecutionID) + runID := strings.TrimSpace(req.Filters.RunID) + if executionID == "" && runID == "" { + respondError(c, http.StatusBadRequest, "missing_filter", + "events queries require filters.execution_id or filters.run_id") + return + } + + var since, until *time.Time + if req.Filters.Since != "" { + if t, err := time.Parse(time.RFC3339, req.Filters.Since); err == nil { + since = &t + } + } + if req.Filters.Until != "" { + if t, err := time.Parse(time.RFC3339, req.Filters.Until); err == nil { + until = &t + } + } + + var executionIDs []string + if executionID != "" { + executionIDs = []string{executionID} + } else { + filter := types.ExecutionFilter{RunID: &runID} + execs, err := store.QueryExecutionRecords(ctx, filter) + if err != nil { + respondError(c, http.StatusInternalServerError, "query_failed", err.Error()) + return + } + executionIDs = make([]string, 0, len(execs)) + for _, exec := range execs { + executionIDs = append(executionIDs, exec.ExecutionID) + } + } + + events := make([]*types.WorkflowExecutionEvent, 0) + for _, id := range executionIDs { + evts, err := store.ListWorkflowExecutionEvents(ctx, id, nil, 0) + if err != nil { + respondError(c, http.StatusInternalServerError, "query_failed", err.Error()) + return + } + for _, evt := range evts { + if since != nil && evt.EmittedAt.Before(*since) { + continue + } + if until != nil && evt.EmittedAt.After(*until) { + continue + } + events = append(events, evt) + } + } + + sort.SliceStable(events, func(i, j int) bool { + if events[i].EmittedAt.Equal(events[j].EmittedAt) { + return events[i].Sequence < events[j].Sequence + } + return events[i].EmittedAt.Before(events[j].EmittedAt) + }) + + total := len(events) + if req.Offset > 0 { + if req.Offset < len(events) { + events = events[req.Offset:] + } else { + events = events[:0] + } + } + if len(events) > req.Limit { + events = events[:req.Limit] + } + + respondOK(c, gin.H{ + "resource": "events", + "results": events, + "total": total, + "limit": req.Limit, + "offset": req.Offset, + }) +} + func querySessions(c *gin.Context, ctx context.Context, store storage.StorageProvider, req QueryRequest) { filters := types.SessionFilters{} if req.Filters.ActorID != "" { diff --git a/control-plane/internal/handlers/agentic/query_events_test.go b/control-plane/internal/handlers/agentic/query_events_test.go new file mode 100644 index 000000000..f70c79314 --- /dev/null +++ b/control-plane/internal/handlers/agentic/query_events_test.go @@ -0,0 +1,256 @@ +package agentic + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/Agent-Field/agentfield/control-plane/pkg/types" + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// eventsTestStorage overrides the two storage calls queryEvents composes. +type eventsTestStorage struct { + *mockStatusStorage + queryExecutionRecordsFn func(context.Context, types.ExecutionFilter) ([]*types.Execution, error) + listWorkflowExecutionEventsFn func(context.Context, string, *int64, int) ([]*types.WorkflowExecutionEvent, error) +} + +func (s *eventsTestStorage) QueryExecutionRecords(ctx context.Context, filter types.ExecutionFilter) ([]*types.Execution, error) { + if s.queryExecutionRecordsFn != nil { + return s.queryExecutionRecordsFn(ctx, filter) + } + return s.mockStatusStorage.QueryExecutionRecords(ctx, filter) +} + +func (s *eventsTestStorage) ListWorkflowExecutionEvents(ctx context.Context, executionID string, afterSeq *int64, limit int) ([]*types.WorkflowExecutionEvent, error) { + if s.listWorkflowExecutionEventsFn != nil { + return s.listWorkflowExecutionEventsFn(ctx, executionID, afterSeq, limit) + } + return s.mockStatusStorage.ListWorkflowExecutionEvents(ctx, executionID, afterSeq, limit) +} + +func postEventsQuery(t *testing.T, store *eventsTestStorage, body map[string]interface{}) *httptest.ResponseRecorder { + t.Helper() + + router := gin.New() + router.POST("/query", QueryHandler(store)) + + encoded, err := json.Marshal(body) + require.NoError(t, err) + + req := httptest.NewRequest(http.MethodPost, "/query", bytes.NewReader(encoded)) + req.Header.Set("Content-Type", "application/json") + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + return rec +} + +func eventAt(executionID string, seq int64, eventType string, emittedAt time.Time) *types.WorkflowExecutionEvent { + return &types.WorkflowExecutionEvent{ + ExecutionID: executionID, + WorkflowID: "wf-1", + Sequence: seq, + EventType: eventType, + EmittedAt: emittedAt, + } +} + +// Contract: an events query without run_id or execution_id is a 400 with a +// structured error, not a full-table scan. +func TestQueryEvents_RequiresRunOrExecutionFilter(t *testing.T) { + store := &eventsTestStorage{mockStatusStorage: &mockStatusStorage{}} + + rec := postEventsQuery(t, store, map[string]interface{}{"resource": "events"}) + + require.Equal(t, http.StatusBadRequest, rec.Code) + resp := decodeEnvelope(t, rec.Body) + require.False(t, resp.OK) + require.NotNil(t, resp.Error) + assert.Equal(t, "missing_filter", resp.Error.Code) +} + +// Contract: filtering by execution_id returns that execution's persisted +// events sorted by emitted_at ascending, with since/until bounds applied. +func TestQueryEvents_ByExecutionID(t *testing.T) { + base := time.Date(2026, 7, 1, 12, 0, 0, 0, time.UTC) + store := &eventsTestStorage{ + mockStatusStorage: &mockStatusStorage{}, + listWorkflowExecutionEventsFn: func(_ context.Context, executionID string, afterSeq *int64, limit int) ([]*types.WorkflowExecutionEvent, error) { + require.Equal(t, "exec-1", executionID) + require.Nil(t, afterSeq) + require.Zero(t, limit) + return []*types.WorkflowExecutionEvent{ + eventAt("exec-1", 1, "execution.started", base), + eventAt("exec-1", 2, "execution.waiting", base.Add(2*time.Minute)), + eventAt("exec-1", 3, "execution.completed", base.Add(10*time.Minute)), + }, nil + }, + } + + rec := postEventsQuery(t, store, map[string]interface{}{ + "resource": "events", + "filters": map[string]interface{}{ + "execution_id": "exec-1", + "since": base.Add(time.Minute).Format(time.RFC3339), + "until": base.Add(5 * time.Minute).Format(time.RFC3339), + }, + }) + + require.Equal(t, http.StatusOK, rec.Code) + resp := decodeEnvelope(t, rec.Body) + require.True(t, resp.OK) + + data := resp.Data.(map[string]interface{}) + assert.Equal(t, "events", data["resource"]) + results := data["results"].([]interface{}) + require.Len(t, results, 1) + first := results[0].(map[string]interface{}) + assert.Equal(t, "execution.waiting", first["event_type"]) + assert.Equal(t, float64(1), data["total"]) +} + +// Contract: filtering by run_id fans out over the run's executions and merges +// their events into one emitted_at-ascending list (ties broken by sequence). +func TestQueryEvents_ByRunIDMergesAndSorts(t *testing.T) { + base := time.Date(2026, 7, 1, 12, 0, 0, 0, time.UTC) + + eventsByExecution := map[string][]*types.WorkflowExecutionEvent{ + "exec-a": { + eventAt("exec-a", 1, "execution.started", base), + eventAt("exec-a", 2, "execution.completed", base.Add(4*time.Minute)), + }, + "exec-b": { + eventAt("exec-b", 1, "execution.started", base.Add(2*time.Minute)), + // Same timestamp as exec-a seq 2 but higher sequence — sorts after. + eventAt("exec-b", 5, "execution.completed", base.Add(4*time.Minute)), + }, + } + + store := &eventsTestStorage{ + mockStatusStorage: &mockStatusStorage{}, + queryExecutionRecordsFn: func(_ context.Context, filter types.ExecutionFilter) ([]*types.Execution, error) { + require.NotNil(t, filter.RunID) + require.Equal(t, "run-1", *filter.RunID) + return []*types.Execution{ + {ExecutionID: "exec-a", RunID: "run-1"}, + {ExecutionID: "exec-b", RunID: "run-1"}, + }, nil + }, + listWorkflowExecutionEventsFn: func(_ context.Context, executionID string, _ *int64, _ int) ([]*types.WorkflowExecutionEvent, error) { + return eventsByExecution[executionID], nil + }, + } + + rec := postEventsQuery(t, store, map[string]interface{}{ + "resource": "events", + "filters": map[string]interface{}{"run_id": "run-1"}, + }) + + require.Equal(t, http.StatusOK, rec.Code) + resp := decodeEnvelope(t, rec.Body) + require.True(t, resp.OK) + + data := resp.Data.(map[string]interface{}) + results := data["results"].([]interface{}) + require.Len(t, results, 4) + assert.Equal(t, float64(4), data["total"]) + + var order []string + for _, raw := range results { + evt := raw.(map[string]interface{}) + order = append(order, evt["execution_id"].(string)) + } + assert.Equal(t, []string{"exec-a", "exec-b", "exec-a", "exec-b"}, order) +} + +// Contract: limit and offset paginate after filtering; total reflects the +// pre-pagination count. +func TestQueryEvents_LimitAndOffset(t *testing.T) { + base := time.Date(2026, 7, 1, 12, 0, 0, 0, time.UTC) + all := []*types.WorkflowExecutionEvent{ + eventAt("exec-1", 1, "e1", base), + eventAt("exec-1", 2, "e2", base.Add(time.Minute)), + eventAt("exec-1", 3, "e3", base.Add(2*time.Minute)), + eventAt("exec-1", 4, "e4", base.Add(3*time.Minute)), + } + store := &eventsTestStorage{ + mockStatusStorage: &mockStatusStorage{}, + listWorkflowExecutionEventsFn: func(_ context.Context, _ string, _ *int64, _ int) ([]*types.WorkflowExecutionEvent, error) { + return all, nil + }, + } + + rec := postEventsQuery(t, store, map[string]interface{}{ + "resource": "events", + "filters": map[string]interface{}{"execution_id": "exec-1"}, + "limit": 2, + "offset": 1, + }) + + require.Equal(t, http.StatusOK, rec.Code) + resp := decodeEnvelope(t, rec.Body) + data := resp.Data.(map[string]interface{}) + results := data["results"].([]interface{}) + require.Len(t, results, 2) + assert.Equal(t, "e2", results[0].(map[string]interface{})["event_type"]) + assert.Equal(t, "e3", results[1].(map[string]interface{})["event_type"]) + assert.Equal(t, float64(4), data["total"]) + + // Offset past the end yields an empty page, not an error. + rec = postEventsQuery(t, store, map[string]interface{}{ + "resource": "events", + "filters": map[string]interface{}{"execution_id": "exec-1"}, + "limit": 2, + "offset": 10, + }) + require.Equal(t, http.StatusOK, rec.Code) + resp = decodeEnvelope(t, rec.Body) + data = resp.Data.(map[string]interface{}) + assert.Empty(t, data["results"]) + assert.Equal(t, float64(4), data["total"]) +} + +// Contract: storage failures surface as 500 query_failed envelopes. +func TestQueryEvents_StorageErrors(t *testing.T) { + t.Run("event listing fails", func(t *testing.T) { + store := &eventsTestStorage{ + mockStatusStorage: &mockStatusStorage{}, + listWorkflowExecutionEventsFn: func(_ context.Context, _ string, _ *int64, _ int) ([]*types.WorkflowExecutionEvent, error) { + return nil, errors.New("db down") + }, + } + rec := postEventsQuery(t, store, map[string]interface{}{ + "resource": "events", + "filters": map[string]interface{}{"execution_id": "exec-1"}, + }) + require.Equal(t, http.StatusInternalServerError, rec.Code) + resp := decodeEnvelope(t, rec.Body) + require.NotNil(t, resp.Error) + assert.Equal(t, "query_failed", resp.Error.Code) + }) + + t.Run("run fan-out fails", func(t *testing.T) { + store := &eventsTestStorage{ + mockStatusStorage: &mockStatusStorage{}, + queryExecutionRecordsFn: func(_ context.Context, _ types.ExecutionFilter) ([]*types.Execution, error) { + return nil, errors.New("db down") + }, + } + rec := postEventsQuery(t, store, map[string]interface{}{ + "resource": "events", + "filters": map[string]interface{}{"run_id": "run-1"}, + }) + require.Equal(t, http.StatusInternalServerError, rec.Code) + resp := decodeEnvelope(t, rec.Body) + require.NotNil(t, resp.Error) + assert.Equal(t, "query_failed", resp.Error.Code) + }) +} From e2c66547f6daa26b7c25e7be6402c04e9c47265e Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Fri, 10 Jul 2026 15:30:45 -0400 Subject: [PATCH 4/9] feat(control-plane): --json envelopes for node lifecycle commands Add a --json flag to af list, af install, af run, af stop, and af logs that emits the agent-mode {ok,data,error:{code,message,hint}} envelope on stdout with non-zero exit on error. Default human output is unchanged. - list: {nodes:[{name,version,status,port,description}], total} - install/run: envelope on the real stdout; service-layer progress prints are redirected to stderr for the duration - stop: {node, status: stopped|not_running}; progress output is suppressed via a Quiet flag on AgentNodeStopper (printf wrapper only, no changes to the signal/liveness logic) - logs: {node, log_path, lines:[...]} honoring --tail via a Go-native tail (no external tail binary); --follow with --json is rejected as invalid_flags since a stream cannot be a JSON snapshot Co-Authored-By: Claude Fable 5 --- .../internal/cli/commands/install.go | 23 ++ .../internal/cli/commands/json_output.go | 55 +++ .../internal/cli/commands/json_output_test.go | 141 +++++++ control-plane/internal/cli/commands/run.go | 36 ++ control-plane/internal/cli/list.go | 58 ++- control-plane/internal/cli/logs.go | 88 ++++- control-plane/internal/cli/node_json.go | 23 ++ control-plane/internal/cli/node_json_test.go | 360 ++++++++++++++++++ control-plane/internal/cli/stop.go | 49 ++- 9 files changed, 820 insertions(+), 13 deletions(-) create mode 100644 control-plane/internal/cli/commands/json_output.go create mode 100644 control-plane/internal/cli/commands/json_output_test.go create mode 100644 control-plane/internal/cli/node_json.go create mode 100644 control-plane/internal/cli/node_json_test.go diff --git a/control-plane/internal/cli/commands/install.go b/control-plane/internal/cli/commands/install.go index 50fe2cf47..0625b5f9d 100644 --- a/control-plane/internal/cli/commands/install.go +++ b/control-plane/internal/cli/commands/install.go @@ -36,6 +36,7 @@ func (cmd *InstallCommand) GetDescription() string { func (cmd *InstallCommand) BuildCobraCommand() *cobra.Command { var force bool var verbose bool + var jsonOutput bool cobraCmd := &cobra.Command{ Use: "install ", @@ -53,6 +54,9 @@ Examples: agentfield install agent-name`, Args: cobra.ExactArgs(1), RunE: func(cobraCmd *cobra.Command, args []string) error { + if jsonOutput { + return cmd.executeJSON(args[0], force) + } // Update output formatter with verbose setting cmd.output.SetVerbose(verbose) return cmd.execute(args[0], force, verbose) @@ -61,10 +65,29 @@ Examples: cobraCmd.Flags().BoolVarP(&force, "force", "f", false, "Force reinstall if package exists") cobraCmd.Flags().BoolVarP(&verbose, "verbose", "v", false, "Verbose output") + cobraCmd.Flags().BoolVar(&jsonOutput, "json", false, "Emit a machine-readable JSON envelope (diagnostics go to stderr)") return cobraCmd } +// executeJSON installs the package and emits a JSON envelope on stdout. +// Service-layer progress output is redirected to stderr for the duration. +func (cmd *InstallCommand) executeJSON(packagePath string, force bool) error { + stdout, restore := redirectStdoutToStderr() + defer restore() + + options := domain.InstallOptions{Force: force} + if err := cmd.Services.PackageService.InstallPackage(packagePath, options); err != nil { + printJSONError(stdout, "install_failed", err.Error(), "Check the package source (path, git URL, or registry name); use --force to reinstall.") + return err + } + + return printJSONSuccess(stdout, map[string]interface{}{ + "source": packagePath, + "status": "installed", + }) +} + // execute performs the actual installation func (cmd *InstallCommand) execute(packagePath string, force, verbose bool) error { cmd.output.PrintHeader("Installing AgentField Package") diff --git a/control-plane/internal/cli/commands/json_output.go b/control-plane/internal/cli/commands/json_output.go new file mode 100644 index 000000000..352fc26f4 --- /dev/null +++ b/control-plane/internal/cli/commands/json_output.go @@ -0,0 +1,55 @@ +package commands + +import ( + "encoding/json" + "fmt" + "os" +) + +// jsonEnvelope mirrors the af agent-mode AgentResponse envelope shape +// ({ok, data, error:{code,message,hint}}) for framework-based lifecycle +// commands (install/run) running with --json. +type jsonEnvelope struct { + OK bool `json:"ok"` + Data interface{} `json:"data,omitempty"` + Error *jsonEnvelopeError `json:"error,omitempty"` +} + +type jsonEnvelopeError struct { + Code string `json:"code"` + Message string `json:"message"` + Hint string `json:"hint,omitempty"` +} + +// printJSONEnvelope writes the envelope to the given writer (the real stdout, +// which callers capture before redirecting service diagnostics to stderr). +func printJSONEnvelope(out *os.File, env jsonEnvelope) error { + b, err := json.MarshalIndent(env, "", " ") + if err != nil { + return fmt.Errorf("marshal output: %w", err) + } + if _, err := fmt.Fprintln(out, string(b)); err != nil { + return fmt.Errorf("write output: %w", err) + } + return nil +} + +func printJSONSuccess(out *os.File, data interface{}) error { + return printJSONEnvelope(out, jsonEnvelope{OK: true, Data: data}) +} + +func printJSONError(out *os.File, code, message, hint string) { + _ = printJSONEnvelope(out, jsonEnvelope{ + OK: false, + Error: &jsonEnvelopeError{Code: code, Message: message, Hint: hint}, + }) +} + +// redirectStdoutToStderr points os.Stdout at os.Stderr so service-layer +// progress prints don't pollute the JSON envelope on the real stdout. It +// returns the original stdout and a restore func. +func redirectStdoutToStderr() (*os.File, func()) { + original := os.Stdout + os.Stdout = os.Stderr + return original, func() { os.Stdout = original } +} diff --git a/control-plane/internal/cli/commands/json_output_test.go b/control-plane/internal/cli/commands/json_output_test.go new file mode 100644 index 000000000..630c36515 --- /dev/null +++ b/control-plane/internal/cli/commands/json_output_test.go @@ -0,0 +1,141 @@ +package commands + +import ( + "encoding/json" + "errors" + "testing" + "time" + + "github.com/Agent-Field/agentfield/control-plane/internal/cli/framework" + "github.com/Agent-Field/agentfield/control-plane/internal/core/domain" + "github.com/stretchr/testify/require" +) + +func decodeJSONEnvelope(t *testing.T, output string) jsonEnvelope { + t.Helper() + + var env jsonEnvelope + require.NoError(t, json.Unmarshal([]byte(output), &env), "stdout must be a single JSON envelope, got: %s", output) + return env +} + +// Contract: `af install --json` emits {ok:true,data:{source,status}} on +// stdout and nothing else. +func TestInstallCommandJSON(t *testing.T) { + pkgSvc := &fakePackageService{} + command := NewInstallCommand(&framework.ServiceContainer{PackageService: pkgSvc}) + + cobraCmd := command.BuildCobraCommand() + cobraCmd.SetArgs([]string{"./my-agent", "--json", "--force"}) + cobraCmd.SilenceUsage = true + cobraCmd.SilenceErrors = true + + var execErr error + output := captureStdout(t, func() { + execErr = cobraCmd.Execute() + }) + + require.NoError(t, execErr) + require.Equal(t, 1, pkgSvc.installCalls) + require.Equal(t, "./my-agent", pkgSvc.lastSource) + require.True(t, pkgSvc.lastOptions.Force) + + env := decodeJSONEnvelope(t, output) + require.True(t, env.OK) + data := env.Data.(map[string]interface{}) + require.Equal(t, "./my-agent", data["source"]) + require.Equal(t, "installed", data["status"]) +} + +// Contract: install failure emits {ok:false,error:{code:install_failed}} and +// returns the error (non-zero exit). +func TestInstallCommandJSONFailure(t *testing.T) { + pkgSvc := &fakePackageService{installErr: errors.New("no such package")} + command := NewInstallCommand(&framework.ServiceContainer{PackageService: pkgSvc}) + + cobraCmd := command.BuildCobraCommand() + cobraCmd.SetArgs([]string{"ghost", "--json"}) + cobraCmd.SilenceUsage = true + cobraCmd.SilenceErrors = true + + var execErr error + output := captureStdout(t, func() { + execErr = cobraCmd.Execute() + }) + + require.Error(t, execErr) + + env := decodeJSONEnvelope(t, output) + require.False(t, env.OK) + require.NotNil(t, env.Error) + require.Equal(t, "install_failed", env.Error.Code) + require.Contains(t, env.Error.Message, "no such package") +} + +// Contract: `af run --json` emits {ok:true,data:{node,pid,port,status, +// log_file,started_at,detach}}. +func TestRunCommandJSON(t *testing.T) { + started := time.Date(2026, 7, 1, 12, 0, 0, 0, time.UTC) + agentSvc := &fakeAgentService{ + runningAgent: &domain.RunningAgent{ + Name: "demo", + PID: 4242, + Port: 8005, + Status: "running", + StartedAt: started, + LogFile: "/tmp/demo.log", + }, + } + command := NewRunCommand(&framework.ServiceContainer{AgentService: agentSvc}) + + cobraCmd := command.BuildCobraCommand() + cobraCmd.SetArgs([]string{"demo", "--json", "--port", "8005"}) + cobraCmd.SilenceUsage = true + cobraCmd.SilenceErrors = true + + var execErr error + output := captureStdout(t, func() { + execErr = cobraCmd.Execute() + }) + + require.NoError(t, execErr) + require.Equal(t, 1, agentSvc.runCalls) + require.Equal(t, "demo", agentSvc.lastName) + require.Equal(t, 8005, agentSvc.lastOptions.Port) + require.True(t, agentSvc.lastOptions.Detach) + + env := decodeJSONEnvelope(t, output) + require.True(t, env.OK) + data := env.Data.(map[string]interface{}) + require.Equal(t, "demo", data["node"]) + require.Equal(t, float64(4242), data["pid"]) + require.Equal(t, float64(8005), data["port"]) + require.Equal(t, "running", data["status"]) + require.Equal(t, "/tmp/demo.log", data["log_file"]) + require.Equal(t, "2026-07-01T12:00:00Z", data["started_at"]) + require.Equal(t, true, data["detach"]) +} + +// Contract: run failure emits {ok:false,error:{code:run_failed}} and returns +// the error. +func TestRunCommandJSONFailure(t *testing.T) { + agentSvc := &fakeAgentService{runErr: errors.New("port in use")} + command := NewRunCommand(&framework.ServiceContainer{AgentService: agentSvc}) + + cobraCmd := command.BuildCobraCommand() + cobraCmd.SetArgs([]string{"demo", "--json"}) + cobraCmd.SilenceUsage = true + cobraCmd.SilenceErrors = true + + var execErr error + output := captureStdout(t, func() { + execErr = cobraCmd.Execute() + }) + + require.Error(t, execErr) + + env := decodeJSONEnvelope(t, output) + require.False(t, env.OK) + require.Equal(t, "run_failed", env.Error.Code) + require.Contains(t, env.Error.Message, "port in use") +} diff --git a/control-plane/internal/cli/commands/run.go b/control-plane/internal/cli/commands/run.go index ad5fe7a18..ad837e984 100644 --- a/control-plane/internal/cli/commands/run.go +++ b/control-plane/internal/cli/commands/run.go @@ -2,6 +2,7 @@ package commands import ( "fmt" + "time" "github.com/Agent-Field/agentfield/control-plane/internal/cli/framework" "github.com/Agent-Field/agentfield/control-plane/internal/core/domain" @@ -37,6 +38,7 @@ func (cmd *RunCommand) BuildCobraCommand() *cobra.Command { var port int var detach bool var verbose bool + var jsonOutput bool cobraCmd := &cobra.Command{ Use: "run ", @@ -52,6 +54,9 @@ Examples: af run my-agent --detach=false`, Args: cobra.ExactArgs(1), RunE: func(cobraCmd *cobra.Command, args []string) error { + if jsonOutput { + return cmd.executeJSON(args[0], port, detach) + } // Update output formatter with verbose setting cmd.output.SetVerbose(verbose) return cmd.execute(args[0], port, detach, verbose) @@ -61,10 +66,41 @@ Examples: cobraCmd.Flags().IntVarP(&port, "port", "p", 0, "Specific port to use (auto-assigned if not specified)") cobraCmd.Flags().BoolVarP(&detach, "detach", "d", true, "Run in background (default: true)") cobraCmd.Flags().BoolVarP(&verbose, "verbose", "v", false, "Verbose output") + cobraCmd.Flags().BoolVar(&jsonOutput, "json", false, "Emit a machine-readable JSON envelope (diagnostics go to stderr)") return cobraCmd } +// executeJSON starts the agent node and emits a JSON envelope on stdout. +// Service-layer progress output is redirected to stderr for the duration. +func (cmd *RunCommand) executeJSON(agentName string, port int, detach bool) error { + stdout, restore := redirectStdoutToStderr() + defer restore() + + options := domain.RunOptions{Port: port, Detach: detach} + runningAgent, err := cmd.Services.AgentService.RunAgent(agentName, options) + if err != nil { + printJSONError(stdout, "run_failed", err.Error(), "Run 'af list --json' to check the node is installed and not already running.") + return err + } + + data := map[string]interface{}{ + "node": agentName, + "pid": runningAgent.PID, + "port": runningAgent.Port, + "status": runningAgent.Status, + "detach": detach, + } + if runningAgent.LogFile != "" { + data["log_file"] = runningAgent.LogFile + } + if !runningAgent.StartedAt.IsZero() { + data["started_at"] = runningAgent.StartedAt.UTC().Format(time.RFC3339) + } + + return printJSONSuccess(stdout, data) +} + // execute performs the actual agent execution func (cmd *RunCommand) execute(agentName string, port int, detach, verbose bool) error { cmd.output.PrintHeader("Running AgentField Agent") diff --git a/control-plane/internal/cli/list.go b/control-plane/internal/cli/list.go index c95e64b45..6b42da686 100644 --- a/control-plane/internal/cli/list.go +++ b/control-plane/internal/cli/list.go @@ -13,6 +13,8 @@ import ( "gopkg.in/yaml.v3" ) +var listJSON bool + // NewListCommand creates the list command func NewListCommand() *cobra.Command { cmd := &cobra.Command{ @@ -23,13 +25,65 @@ func NewListCommand() *cobra.Command { Shows package name, version, status (running/stopped), and port if running. Examples: - af list`, - Run: runListCommand, + af list + af list --json`, + RunE: func(cmd *cobra.Command, args []string) error { + if listJSON { + return runListCommandJSON() + } + runListCommand(cmd, args) + return nil + }, } + cmd.Flags().BoolVar(&listJSON, "json", false, "Emit a machine-readable JSON envelope instead of the table") return cmd } +// runListCommandJSON emits the installed-node registry as a JSON envelope. +func runListCommandJSON() error { + agentfieldHome := getAgentFieldHomeDir() + registryPath := filepath.Join(agentfieldHome, "installed.yaml") + + registry := &packages.InstallationRegistry{ + Installed: make(map[string]packages.InstalledPackage), + } + + if data, err := os.ReadFile(registryPath); err == nil { + if err := yaml.Unmarshal(data, registry); err != nil { + return nodeJSONError("registry_error", fmt.Sprintf("failed to parse registry: %v", err), "Inspect "+registryPath+" for corruption.") + } + } else if !errors.Is(err, os.ErrNotExist) { + return nodeJSONError("registry_error", fmt.Sprintf("failed to read registry: %v", err), "Check permissions on "+registryPath+".") + } + + names := make([]string, 0, len(registry.Installed)) + for name := range registry.Installed { + names = append(names, name) + } + sort.Strings(names) + + nodes := make([]map[string]interface{}, 0, len(names)) + for _, name := range names { + pkg := registry.Installed[name] + node := map[string]interface{}{ + "name": name, + "version": pkg.Version, + "status": pkg.Status, + "description": pkg.Description, + } + if pkg.Status == "running" && pkg.Runtime.Port != nil { + node["port"] = *pkg.Runtime.Port + } + nodes = append(nodes, node) + } + + return nodeJSONSuccess(map[string]interface{}{ + "nodes": nodes, + "total": len(nodes), + }) +} + func runListCommand(cmd *cobra.Command, args []string) { agentfieldHome := getAgentFieldHomeDir() registryPath := filepath.Join(agentfieldHome, "installed.yaml") diff --git a/control-plane/internal/cli/logs.go b/control-plane/internal/cli/logs.go index c08917c76..662c8ec4e 100644 --- a/control-plane/internal/cli/logs.go +++ b/control-plane/internal/cli/logs.go @@ -6,6 +6,7 @@ import ( "os" "os/exec" // Added missing import "path/filepath" + "strings" "github.com/Agent-Field/agentfield/control-plane/internal/logger" "github.com/Agent-Field/agentfield/control-plane/internal/packages" @@ -16,6 +17,7 @@ import ( var ( logsFollow bool logsTail int + logsJSON bool ) // NewLogsCommand creates the logs command @@ -29,13 +31,15 @@ Shows the most recent log entries from the agent node's log file. Examples: af logs email-helper - af logs data-analyzer --follow`, + af logs data-analyzer --follow + af logs email-helper --json --tail 100`, Args: cobra.ExactArgs(1), RunE: runLogsCommand, } cmd.Flags().BoolVarP(&logsFollow, "follow", "f", false, "Follow log output") cmd.Flags().IntVarP(&logsTail, "tail", "n", 50, "Number of lines to show from the end") + cmd.Flags().BoolVar(&logsJSON, "json", false, "Emit a machine-readable JSON envelope ({node, log_path, lines}) instead of raw log text") return cmd } @@ -43,6 +47,10 @@ Examples: func runLogsCommand(cmd *cobra.Command, args []string) error { agentNodeName := args[0] + if logsJSON { + return runLogsCommandJSON(agentNodeName) + } + logViewer := &LogViewer{ AgentFieldHome: getAgentFieldHomeDir(), Follow: logsFollow, @@ -57,6 +65,30 @@ func runLogsCommand(cmd *cobra.Command, args []string) error { return nil } +// runLogsCommandJSON emits the last --tail log lines as a JSON envelope. +// --follow is a live stream and cannot be combined with a JSON snapshot. +func runLogsCommandJSON(agentNodeName string) error { + if logsFollow { + return nodeJSONError("invalid_flags", "--follow cannot be combined with --json", "Poll 'af logs --json --tail N' instead of following.") + } + + logViewer := &LogViewer{ + AgentFieldHome: getAgentFieldHomeDir(), + Tail: logsTail, + } + + logPath, lines, err := logViewer.CollectLogLines(agentNodeName) + if err != nil { + return nodeJSONError("logs_failed", err.Error(), "Run 'af list --json' to see installed nodes.") + } + + return nodeJSONSuccess(map[string]interface{}{ + "node": agentNodeName, + "log_path": logPath, + "lines": lines, + }) +} + // LogViewer handles viewing agent node logs type LogViewer struct { AgentFieldHome string @@ -102,6 +134,60 @@ func (lv *LogViewer) ViewLogs(agentNodeName string) error { } } +// CollectLogLines returns the node's log file path and its last Tail lines +// for structured (JSON) output. A missing log file yields empty lines rather +// than an error, mirroring the human command's behavior for nodes that have +// not run yet. +func (lv *LogViewer) CollectLogLines(agentNodeName string) (string, []string, error) { + registryPath := filepath.Join(lv.AgentFieldHome, "installed.yaml") + registry := &packages.InstallationRegistry{ + Installed: make(map[string]packages.InstalledPackage), + } + + if data, err := os.ReadFile(registryPath); err == nil { + if err := yaml.Unmarshal(data, registry); err != nil { + return "", nil, fmt.Errorf("failed to parse registry: %w", err) + } + } else if !errors.Is(err, os.ErrNotExist) { + return "", nil, fmt.Errorf("failed to read registry: %w", err) + } + + agentNode, exists := registry.Installed[agentNodeName] + if !exists { + return "", nil, fmt.Errorf("agent node %s not installed", agentNodeName) + } + + logFile := agentNode.Runtime.LogFile + if _, err := os.Stat(logFile); os.IsNotExist(err) { + return logFile, []string{}, nil + } + + lines, err := tailFileLines(logFile, lv.Tail) + if err != nil { + return logFile, nil, fmt.Errorf("failed to read log file: %w", err) + } + return logFile, lines, nil +} + +// tailFileLines returns the last n lines of the file at path. +func tailFileLines(path string, n int) ([]string, error) { + data, err := os.ReadFile(path) + if err != nil { + return nil, err + } + + text := strings.TrimRight(string(data), "\n") + if text == "" { + return []string{}, nil + } + + lines := strings.Split(text, "\n") + if n > 0 && len(lines) > n { + lines = lines[len(lines)-n:] + } + return lines, nil +} + // tailLogs shows the last N lines of the log file func (lv *LogViewer) tailLogs(logFile string, lines int) error { cmd := exec.Command("tail", "-n", fmt.Sprintf("%d", lines), logFile) diff --git a/control-plane/internal/cli/node_json.go b/control-plane/internal/cli/node_json.go new file mode 100644 index 000000000..ccf443c43 --- /dev/null +++ b/control-plane/internal/cli/node_json.go @@ -0,0 +1,23 @@ +package cli + +import "errors" + +// Node lifecycle commands (af list/stop/logs) gain a --json flag that emits +// the same {ok,data,error:{code,message,hint}} envelope agent mode uses. +// Default human output is unchanged; --json guarantees stdout carries exactly +// one JSON envelope and errors exit non-zero. + +// nodeJSONSuccess emits an {ok:true,data:...} envelope on stdout. +func nodeJSONSuccess(data interface{}) error { + return outputAgentJSON(AgentResponse{OK: true, Data: data}) +} + +// nodeJSONError emits an {ok:false,error:{...}} envelope on stdout and returns +// a cliExitError so the process exits non-zero with the message on stderr. +func nodeJSONError(code, message, hint string) error { + _ = outputAgentJSON(AgentResponse{ + OK: false, + Error: &AgentError{Code: code, Message: message, Hint: hint}, + }) + return cliExitError{Code: 1, Err: errors.New(message)} +} diff --git a/control-plane/internal/cli/node_json_test.go b/control-plane/internal/cli/node_json_test.go new file mode 100644 index 000000000..978ec56b0 --- /dev/null +++ b/control-plane/internal/cli/node_json_test.go @@ -0,0 +1,360 @@ +package cli + +import ( + "encoding/json" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/Agent-Field/agentfield/control-plane/internal/packages" + "github.com/stretchr/testify/require" + "gopkg.in/yaml.v3" +) + +// resetLifecycleFlags restores the package-level --json flag state so tests +// that call run*Command functions directly (without re-registering flags via +// New*Command) are unaffected by ordering. +func resetLifecycleFlags(t *testing.T) { + t.Helper() + t.Cleanup(func() { + listJSON = false + stopJSON = false + logsJSON = false + logsFollow = false + logsTail = 50 + }) +} + +func writeRegistry(t *testing.T, home string, registry *packages.InstallationRegistry) { + t.Helper() + + data, err := yaml.Marshal(registry) + require.NoError(t, err) + require.NoError(t, os.WriteFile(filepath.Join(home, "installed.yaml"), data, 0o644)) +} + +func decodeNodeEnvelope(t *testing.T, output string) AgentResponse { + t.Helper() + + var resp AgentResponse + require.NoError(t, json.Unmarshal([]byte(output), &resp), "stdout must be a single JSON envelope, got: %s", output) + return resp +} + +// Contract: `af list --json` emits {ok:true,data:{nodes:[...],total}} with +// port only for running nodes, and nothing but the envelope on stdout. +func TestListCommandJSON(t *testing.T) { + resetLifecycleFlags(t) + home := t.TempDir() + t.Setenv("AGENTFIELD_HOME", home) + + port := 8005 + writeRegistry(t, home, &packages.InstallationRegistry{ + Installed: map[string]packages.InstalledPackage{ + "beta-node": { + Name: "beta-node", + Version: "0.2.0", + Description: "second node", + Status: "stopped", + }, + "alpha-node": { + Name: "alpha-node", + Version: "1.0.0", + Description: "first node", + Status: "running", + Runtime: packages.RuntimeInfo{Port: &port}, + }, + }, + }) + + output := captureOutput(t, func() { + cmd := NewListCommand() + cmd.SetArgs([]string{"--json"}) + require.NoError(t, cmd.Execute()) + }) + + resp := decodeNodeEnvelope(t, output) + require.True(t, resp.OK) + + data := resp.Data.(map[string]interface{}) + require.Equal(t, float64(2), data["total"]) + nodes := data["nodes"].([]interface{}) + require.Len(t, nodes, 2) + + first := nodes[0].(map[string]interface{}) + require.Equal(t, "alpha-node", first["name"]) + require.Equal(t, "1.0.0", first["version"]) + require.Equal(t, "running", first["status"]) + require.Equal(t, float64(8005), first["port"]) + + second := nodes[1].(map[string]interface{}) + require.Equal(t, "beta-node", second["name"]) + require.Equal(t, "stopped", second["status"]) + _, hasPort := second["port"] + require.False(t, hasPort) +} + +// Contract: an empty registry is a success with zero nodes, not an error. +func TestListCommandJSONEmpty(t *testing.T) { + resetLifecycleFlags(t) + home := t.TempDir() + t.Setenv("AGENTFIELD_HOME", home) + + output := captureOutput(t, func() { + cmd := NewListCommand() + cmd.SetArgs([]string{"--json"}) + require.NoError(t, cmd.Execute()) + }) + + resp := decodeNodeEnvelope(t, output) + require.True(t, resp.OK) + data := resp.Data.(map[string]interface{}) + require.Equal(t, float64(0), data["total"]) +} + +// Contract: a corrupt registry is {ok:false,error:{code:registry_error}} with +// a non-zero exit (cliExitError). +func TestListCommandJSONCorruptRegistry(t *testing.T) { + resetLifecycleFlags(t) + home := t.TempDir() + t.Setenv("AGENTFIELD_HOME", home) + require.NoError(t, os.WriteFile(filepath.Join(home, "installed.yaml"), []byte("installed: ["), 0o644)) + + var execErr error + output := captureOutput(t, func() { + cmd := NewListCommand() + cmd.SetArgs([]string{"--json"}) + cmd.SilenceErrors = true + execErr = cmd.Execute() + }) + + require.Error(t, execErr) + require.True(t, IsCLIExitError(execErr)) + require.Equal(t, 1, ExitCode(execErr)) + + resp := decodeNodeEnvelope(t, output) + require.False(t, resp.OK) + require.NotNil(t, resp.Error) + require.Equal(t, "registry_error", resp.Error.Code) +} + +// Contract: stopping a node that is not running is a no-op success reporting +// status not_running, with no human progress text on stdout. +func TestStopCommandJSONNotRunning(t *testing.T) { + resetLifecycleFlags(t) + home := t.TempDir() + t.Setenv("AGENTFIELD_HOME", home) + + writeRegistry(t, home, &packages.InstallationRegistry{ + Installed: map[string]packages.InstalledPackage{ + "demo": {Name: "demo", Version: "1.0.0", Status: "stopped"}, + }, + }) + + output := captureOutput(t, func() { + cmd := NewStopCommand() + cmd.SetArgs([]string{"demo", "--json"}) + require.NoError(t, cmd.Execute()) + }) + + resp := decodeNodeEnvelope(t, output) + require.True(t, resp.OK) + data := resp.Data.(map[string]interface{}) + require.Equal(t, "demo", data["node"]) + require.Equal(t, "not_running", data["status"]) + require.NotContains(t, output, "is not running") // no human progress line +} + +// Contract: stopping a node that is not installed is a structured error with +// non-zero exit. +func TestStopCommandJSONNotInstalled(t *testing.T) { + resetLifecycleFlags(t) + home := t.TempDir() + t.Setenv("AGENTFIELD_HOME", home) + + var execErr error + output := captureOutput(t, func() { + cmd := NewStopCommand() + cmd.SetArgs([]string{"ghost", "--json"}) + cmd.SilenceErrors = true + execErr = cmd.Execute() + }) + + require.Error(t, execErr) + require.True(t, IsCLIExitError(execErr)) + + resp := decodeNodeEnvelope(t, output) + require.False(t, resp.OK) + require.Equal(t, "stop_failed", resp.Error.Code) + require.Contains(t, resp.Error.Message, "not installed") +} + +// Contract: default (non-json) stop output for a not-running node is +// unchanged. +func TestStopCommandDefaultOutputUnchanged(t *testing.T) { + resetLifecycleFlags(t) + home := t.TempDir() + t.Setenv("AGENTFIELD_HOME", home) + + writeRegistry(t, home, &packages.InstallationRegistry{ + Installed: map[string]packages.InstalledPackage{ + "demo": {Name: "demo", Version: "1.0.0", Status: "stopped"}, + }, + }) + + output := captureOutput(t, func() { + cmd := NewStopCommand() + cmd.SetArgs([]string{"demo"}) + require.NoError(t, cmd.Execute()) + }) + require.Equal(t, "⚠️ Agent node demo is not running\n", output) +} + +// Contract: `af logs --json` returns {node, log_path, lines} honoring --tail. +func TestLogsCommandJSON(t *testing.T) { + resetLifecycleFlags(t) + home := t.TempDir() + t.Setenv("AGENTFIELD_HOME", home) + + require.NoError(t, os.MkdirAll(filepath.Join(home, "logs"), 0o755)) + logFile := filepath.Join(home, "logs", "demo.log") + require.NoError(t, os.WriteFile(logFile, []byte("one\ntwo\nthree\nfour\n"), 0o644)) + + writeRegistry(t, home, &packages.InstallationRegistry{ + Installed: map[string]packages.InstalledPackage{ + "demo": { + Name: "demo", + Version: "1.0.0", + Status: "running", + Runtime: packages.RuntimeInfo{LogFile: logFile}, + }, + }, + }) + + output := captureOutput(t, func() { + cmd := NewLogsCommand() + cmd.SetArgs([]string{"demo", "--json", "--tail", "2"}) + require.NoError(t, cmd.Execute()) + }) + + resp := decodeNodeEnvelope(t, output) + require.True(t, resp.OK) + data := resp.Data.(map[string]interface{}) + require.Equal(t, "demo", data["node"]) + require.Equal(t, logFile, data["log_path"]) + lines := data["lines"].([]interface{}) + require.Equal(t, []interface{}{"three", "four"}, lines) +} + +// Contract: a node that has never run (no log file) yields empty lines, not an +// error — mirroring the human command. +func TestLogsCommandJSONMissingLogFile(t *testing.T) { + resetLifecycleFlags(t) + home := t.TempDir() + t.Setenv("AGENTFIELD_HOME", home) + + writeRegistry(t, home, &packages.InstallationRegistry{ + Installed: map[string]packages.InstalledPackage{ + "demo": { + Name: "demo", + Version: "1.0.0", + Status: "stopped", + Runtime: packages.RuntimeInfo{LogFile: filepath.Join(home, "logs", "never.log")}, + }, + }, + }) + + output := captureOutput(t, func() { + cmd := NewLogsCommand() + cmd.SetArgs([]string{"demo", "--json"}) + require.NoError(t, cmd.Execute()) + }) + + resp := decodeNodeEnvelope(t, output) + require.True(t, resp.OK) + data := resp.Data.(map[string]interface{}) + require.Empty(t, data["lines"]) +} + +// Contract: --follow and --json are mutually exclusive. +func TestLogsCommandJSONFollowRejected(t *testing.T) { + resetLifecycleFlags(t) + home := t.TempDir() + t.Setenv("AGENTFIELD_HOME", home) + + var execErr error + output := captureOutput(t, func() { + cmd := NewLogsCommand() + cmd.SetArgs([]string{"demo", "--json", "--follow"}) + cmd.SilenceErrors = true + execErr = cmd.Execute() + }) + + require.Error(t, execErr) + resp := decodeNodeEnvelope(t, output) + require.False(t, resp.OK) + require.Equal(t, "invalid_flags", resp.Error.Code) +} + +// Contract: logs for an uninstalled node is a structured error. +func TestLogsCommandJSONNotInstalled(t *testing.T) { + resetLifecycleFlags(t) + home := t.TempDir() + t.Setenv("AGENTFIELD_HOME", home) + + var execErr error + output := captureOutput(t, func() { + cmd := NewLogsCommand() + cmd.SetArgs([]string{"ghost", "--json"}) + cmd.SilenceErrors = true + execErr = cmd.Execute() + }) + + require.Error(t, execErr) + resp := decodeNodeEnvelope(t, output) + require.False(t, resp.OK) + require.Equal(t, "logs_failed", resp.Error.Code) + require.Contains(t, resp.Error.Message, "not installed") +} + +func TestTailFileLines(t *testing.T) { + dir := t.TempDir() + + path := filepath.Join(dir, "log.txt") + require.NoError(t, os.WriteFile(path, []byte("a\nb\nc\n"), 0o644)) + + lines, err := tailFileLines(path, 2) + require.NoError(t, err) + require.Equal(t, []string{"b", "c"}, lines) + + lines, err = tailFileLines(path, 10) + require.NoError(t, err) + require.Equal(t, []string{"a", "b", "c"}, lines) + + empty := filepath.Join(dir, "empty.txt") + require.NoError(t, os.WriteFile(empty, nil, 0o644)) + lines, err = tailFileLines(empty, 5) + require.NoError(t, err) + require.Empty(t, lines) + + _, err = tailFileLines(filepath.Join(dir, "missing.txt"), 5) + require.Error(t, err) +} + +// Guard: the JSON envelope helpers must not leak table/panel text into stdout. +func TestListCommandJSONStdoutIsPureJSON(t *testing.T) { + resetLifecycleFlags(t) + home := t.TempDir() + t.Setenv("AGENTFIELD_HOME", home) + + output := captureOutput(t, func() { + cmd := NewListCommand() + cmd.SetArgs([]string{"--json"}) + require.NoError(t, cmd.Execute()) + }) + + trimmed := strings.TrimSpace(output) + require.True(t, strings.HasPrefix(trimmed, "{"), "stdout must start with JSON, got: %q", trimmed) + require.True(t, strings.HasSuffix(trimmed, "}"), "stdout must end with JSON, got: %q", trimmed) +} diff --git a/control-plane/internal/cli/stop.go b/control-plane/internal/cli/stop.go index 6e96f3268..cd77cde7a 100644 --- a/control-plane/internal/cli/stop.go +++ b/control-plane/internal/cli/stop.go @@ -15,6 +15,8 @@ import ( "gopkg.in/yaml.v3" ) +var stopJSON bool + // NewStopCommand creates the stop command func NewStopCommand() *cobra.Command { cmd := &cobra.Command{ @@ -27,11 +29,13 @@ will be updated in the registry. Examples: af stop email-helper - af stop data-analyzer`, + af stop data-analyzer + af stop email-helper --json`, Args: cobra.ExactArgs(1), RunE: runStopCommand, } + cmd.Flags().BoolVar(&stopJSON, "json", false, "Emit a machine-readable JSON envelope instead of progress output") return cmd } @@ -40,18 +44,41 @@ func runStopCommand(cmd *cobra.Command, args []string) error { stopper := &AgentNodeStopper{ AgentFieldHome: getAgentFieldHomeDir(), + Quiet: stopJSON, } if err := stopper.StopAgentNode(agentNodeName); err != nil { + if stopJSON { + return nodeJSONError("stop_failed", err.Error(), "Run 'af list --json' to see installed nodes and their status.") + } return fmt.Errorf("failed to stop agent node: %w", err) } + if stopJSON { + return nodeJSONSuccess(map[string]interface{}{ + "node": agentNodeName, + "status": stopper.Outcome, + }) + } return nil } // AgentNodeStopper handles stopping agent nodes type AgentNodeStopper struct { AgentFieldHome string + // Quiet suppresses human progress output (used by --json mode). + Quiet bool + // Outcome records the result of the last StopAgentNode call: + // "stopped" or "not_running". + Outcome string +} + +// printf writes human progress output unless the stopper is in quiet mode. +func (as *AgentNodeStopper) printf(format string, args ...interface{}) { + if as.Quiet { + return + } + fmt.Printf(format, args...) } // StopAgentNode stops a running agent node @@ -68,7 +95,8 @@ func (as *AgentNodeStopper) StopAgentNode(agentNodeName string) error { } if agentNode.Status != "running" { - fmt.Printf("⚠️ Agent node %s is not running\n", agentNodeName) + as.Outcome = "not_running" + as.printf("⚠️ Agent node %s is not running\n", agentNodeName) return nil } @@ -76,7 +104,7 @@ func (as *AgentNodeStopper) StopAgentNode(agentNodeName string) error { return fmt.Errorf("no PID found for agent node %s", agentNodeName) } - fmt.Printf("🛑 Stopping agent node: %s (PID: %d)\n", agentNodeName, *agentNode.Runtime.PID) + as.printf("🛑 Stopping agent node: %s (PID: %d)\n", agentNodeName, *agentNode.Runtime.PID) // Find and kill the process process, err := os.FindProcess(*agentNode.Runtime.PID) @@ -87,7 +115,7 @@ func (as *AgentNodeStopper) StopAgentNode(agentNodeName string) error { // Try HTTP shutdown first if port is available httpShutdownSuccess := false if agentNode.Runtime.Port != nil { - fmt.Printf("🛑 Attempting graceful HTTP shutdown for agent %s on port %d\n", agentNodeName, *agentNode.Runtime.Port) + as.printf("🛑 Attempting graceful HTTP shutdown for agent %s on port %d\n", agentNodeName, *agentNode.Runtime.Port) // Construct agent base URL baseURL := fmt.Sprintf("http://localhost:%d", *agentNode.Runtime.Port) @@ -111,16 +139,16 @@ func (as *AgentNodeStopper) StopAgentNode(agentNodeName string) error { if err == nil { defer resp.Body.Close() if resp.StatusCode == 200 { - fmt.Printf("✅ HTTP shutdown request accepted for agent %s\n", agentNodeName) + as.printf("✅ HTTP shutdown request accepted for agent %s\n", agentNodeName) httpShutdownSuccess = true // Wait a moment for graceful shutdown time.Sleep(3 * time.Second) } else { - fmt.Printf("⚠️ HTTP shutdown returned status %d for agent %s\n", resp.StatusCode, agentNodeName) + as.printf("⚠️ HTTP shutdown returned status %d for agent %s\n", resp.StatusCode, agentNodeName) } } else { - fmt.Printf("⚠️ HTTP shutdown request failed for agent %s: %v\n", agentNodeName, err) + as.printf("⚠️ HTTP shutdown request failed for agent %s: %v\n", agentNodeName, err) } } } @@ -128,7 +156,7 @@ func (as *AgentNodeStopper) StopAgentNode(agentNodeName string) error { // If HTTP shutdown failed or not available, fall back to process signals if !httpShutdownSuccess { - fmt.Printf("🔄 Falling back to process signal shutdown for agent %s\n", agentNodeName) + as.printf("🔄 Falling back to process signal shutdown for agent %s\n", agentNodeName) // Send SIGTERM for graceful shutdown if err := process.Signal(os.Interrupt); err != nil { @@ -143,7 +171,7 @@ func (as *AgentNodeStopper) StopAgentNode(agentNodeName string) error { // Check if process is still running if err := process.Signal(syscall.Signal(0)); err == nil { // Process still running, force kill - fmt.Printf("⚠️ Process still running, force killing agent %s\n", agentNodeName) + as.printf("⚠️ Process still running, force killing agent %s\n", agentNodeName) if err := process.Kill(); err != nil { return fmt.Errorf("failed to force kill process: %w", err) } @@ -162,7 +190,8 @@ func (as *AgentNodeStopper) StopAgentNode(agentNodeName string) error { return fmt.Errorf("failed to update registry: %w", err) } - fmt.Printf("✅ Agent node %s stopped successfully\n", agentNodeName) + as.Outcome = "stopped" + as.printf("✅ Agent node %s stopped successfully\n", agentNodeName) return nil } From cfac3b452106a91072f230ebea8c56b1af64ae2c Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Fri, 10 Jul 2026 15:43:39 -0400 Subject: [PATCH 5/9] fix(control-plane): persist recorded_at on workflow execution events The raw INSERT in storeWorkflowExecutionEventTx omitted recorded_at (the GORM model's autoCreateTime does not apply to raw SQL), leaving NULL in SQLite/Postgres. ListWorkflowExecutionEvents scanned the column into a non-nullable time.Time, so listing any stored event failed with 'unsupported Scan, storing driver.Value type '. Nothing exercised this listing path end-to-end until the agentic events query resource; the existing storage test masked the bug with a manual UPDATE of recorded_at before listing. Write recorded_at in the INSERT and scan it through sql.NullTime, falling back to emitted_at for legacy NULL rows so old databases keep listing cleanly. Found via runtime verification of 'af agent query -r events' against a local control plane. Co-Authored-By: Claude Fable 5 --- .../coverage_identity_events_reasoner_test.go | 14 +++++++++-- control-plane/internal/storage/local.go | 25 ++++++++++++++----- 2 files changed, 31 insertions(+), 8 deletions(-) diff --git a/control-plane/internal/storage/coverage_identity_events_reasoner_test.go b/control-plane/internal/storage/coverage_identity_events_reasoner_test.go index b2af93d3c..e5174fb0e 100644 --- a/control-plane/internal/storage/coverage_identity_events_reasoner_test.go +++ b/control-plane/internal/storage/coverage_identity_events_reasoner_test.go @@ -163,19 +163,29 @@ func TestExecutionEventAndLogCoverage(t *testing.T) { require.NoError(t, ls.StoreWorkflowExecutionEvent(ctx, event)) require.NotZero(t, event.EventID) require.False(t, event.RecordedAt.IsZero()) - _, err = ls.db.ExecContext(ctx, `UPDATE workflow_execution_events SET recorded_at = ? WHERE event_id = ?`, event.RecordedAt, event.EventID) - require.NoError(t, err) + // The insert must persist recorded_at itself — listing previously + // failed with a NULL-scan error because the raw INSERT omitted it. stored, err := ls.ListWorkflowExecutionEvents(ctx, "exec-events-1", nil, 10) require.NoError(t, err) require.Len(t, stored, 1) require.Equal(t, json.RawMessage("{}"), stored[0].Payload) require.Equal(t, &status, stored[0].Status) + require.False(t, stored[0].RecordedAt.IsZero()) after := int64(1) stored, err = ls.ListWorkflowExecutionEvents(ctx, "exec-events-1", &after, 10) require.NoError(t, err) require.Empty(t, stored) + + // Legacy rows written before recorded_at was persisted have NULL; + // listing must fall back to emitted_at instead of erroring. + _, err = ls.db.ExecContext(ctx, `UPDATE workflow_execution_events SET recorded_at = NULL WHERE event_id = ?`, event.EventID) + require.NoError(t, err) + stored, err = ls.ListWorkflowExecutionEvents(ctx, "exec-events-1", nil, 10) + require.NoError(t, err) + require.Len(t, stored, 1) + require.Equal(t, emittedAt, stored[0].RecordedAt.UTC().Truncate(time.Second)) }) t.Run("stores lists and prunes execution logs", func(t *testing.T) { diff --git a/control-plane/internal/storage/local.go b/control-plane/internal/storage/local.go index 6fbb3139a..99f303b1f 100644 --- a/control-plane/internal/storage/local.go +++ b/control-plane/internal/storage/local.go @@ -7654,11 +7654,17 @@ func (ls *LocalStorage) storeWorkflowExecutionEventTx(ctx context.Context, tx DB payload = "{}" } + // Persist recorded_at explicitly: the GORM model's autoCreateTime does + // not apply to this raw INSERT, and a NULL recorded_at breaks listing. + if event.RecordedAt.IsZero() { + event.RecordedAt = time.Now().UTC() + } + query := ` INSERT INTO workflow_execution_events ( execution_id, workflow_id, run_id, parent_execution_id, sequence, previous_sequence, - event_type, status, status_reason, payload, emitted_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)` + event_type, status, status_reason, payload, emitted_at, recorded_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)` result, err := tx.ExecContext(ctx, query, event.ExecutionID, @@ -7672,6 +7678,7 @@ func (ls *LocalStorage) storeWorkflowExecutionEventTx(ctx context.Context, tx DB event.StatusReason, payload, event.EmittedAt, + event.RecordedAt, ) if err != nil { return fmt.Errorf("failed to insert workflow execution event: %w", err) @@ -7680,9 +7687,6 @@ func (ls *LocalStorage) storeWorkflowExecutionEventTx(ctx context.Context, tx DB if id, err := result.LastInsertId(); err == nil { event.EventID = id } - if event.RecordedAt.IsZero() { - event.RecordedAt = time.Now().UTC() - } return nil } @@ -7720,6 +7724,7 @@ func (ls *LocalStorage) ListWorkflowExecutionEvents(ctx context.Context, executi var status sql.NullString var statusReason sql.NullString var payload sql.NullString + var recordedAt sql.NullTime if err := rows.Scan( &evt.EventID, @@ -7734,11 +7739,19 @@ func (ls *LocalStorage) ListWorkflowExecutionEvents(ctx context.Context, executi &statusReason, &payload, &evt.EmittedAt, - &evt.RecordedAt, + &recordedAt, ); err != nil { return nil, fmt.Errorf("failed to scan workflow execution event: %w", err) } + // Rows written before recorded_at was persisted have NULL here; + // fall back to emitted_at rather than failing the whole listing. + if recordedAt.Valid { + evt.RecordedAt = recordedAt.Time + } else { + evt.RecordedAt = evt.EmittedAt + } + if runID.Valid { evt.RunID = &runID.String } From 72f8d9b1ae986e09e92a56683c4b81a4c9cb84d5 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Fri, 10 Jul 2026 15:49:01 -0400 Subject: [PATCH 6/9] test(control-plane): in-process unit tests for proxy error normalization structuredErrorFromString and defaultCodeForStatus were exercised only through subprocess exit-path tests, which record no coverage. Add direct table-driven tests so the patch-coverage gate sees these lines. Co-Authored-By: Claude Fable 5 --- control-plane/internal/cli/agent_exec_test.go | 42 +++++++++++++++++++ 1 file changed, 42 insertions(+) diff --git a/control-plane/internal/cli/agent_exec_test.go b/control-plane/internal/cli/agent_exec_test.go index 0988aa704..378af8900 100644 --- a/control-plane/internal/cli/agent_exec_test.go +++ b/control-plane/internal/cli/agent_exec_test.go @@ -165,6 +165,48 @@ func TestAgentQueryEventsSendsFilters(t *testing.T) { require.Equal(t, float64(5), payload["limit"]) } +// Contract: legacy string errors normalize into {code,message,hint} — the +// error string becomes the code only when a sibling message carries the +// human detail; otherwise the code derives from the HTTP status. +func TestStructuredErrorFromString(t *testing.T) { + t.Run("bare string error derives code from status", func(t *testing.T) { + got := structuredErrorFromString(map[string]interface{}{}, "execution exec_x not found", http.StatusNotFound) + require.Equal(t, "not_found", got["code"]) + require.Equal(t, "execution exec_x not found", got["message"]) + require.NotEmpty(t, got["hint"]) + }) + + t.Run("error code with sibling message keeps the code", func(t *testing.T) { + payload := map[string]interface{}{"message": "execution is in 'running' state"} + got := structuredErrorFromString(payload, "invalid_state", http.StatusConflict) + require.Equal(t, "invalid_state", got["code"]) + require.Equal(t, "execution is in 'running' state", got["message"]) + }) + + t.Run("blank sibling message falls back to derived code", func(t *testing.T) { + payload := map[string]interface{}{"message": " "} + got := structuredErrorFromString(payload, "boom", http.StatusBadRequest) + require.Equal(t, "bad_request", got["code"]) + require.Equal(t, "boom", got["message"]) + }) +} + +func TestDefaultCodeForStatus(t *testing.T) { + cases := map[int]string{ + http.StatusBadRequest: "bad_request", + http.StatusUnauthorized: "unauthorized", + http.StatusForbidden: "forbidden", + http.StatusNotFound: "not_found", + http.StatusConflict: "conflict", + http.StatusInternalServerError: "server_error", + http.StatusBadGateway: "server_error", + http.StatusTeapot: "request_failed", + } + for status, want := range cases { + require.Equal(t, want, defaultCodeForStatus(status), "status %d", status) + } +} + // Contract: exec subcommand without a verb lists the available verbs. func TestAgentExecListsVerbs(t *testing.T) { oldFormat := outputFormat From a2899a383688dea946e4e3da0892b73b8387edff Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Fri, 10 Jul 2026 15:51:02 -0400 Subject: [PATCH 7/9] feat(control-plane): list approval-response endpoint in discover catalog Register POST /api/v1/executions/:execution_id/approval-response in the agentic API catalog so 'af agent discover' surfaces the new resolution endpoint alongside the other approval routes. Co-Authored-By: Claude Fable 5 --- control-plane/internal/server/apicatalog/catalog_entries.go | 1 + 1 file changed, 1 insertion(+) diff --git a/control-plane/internal/server/apicatalog/catalog_entries.go b/control-plane/internal/server/apicatalog/catalog_entries.go index 42719dd3b..3b21c36c8 100644 --- a/control-plane/internal/server/apicatalog/catalog_entries.go +++ b/control-plane/internal/server/apicatalog/catalog_entries.go @@ -83,6 +83,7 @@ func DefaultEntries() []EndpointEntry { // --- Approval --- {Method: "POST", Path: "/api/v1/executions/:execution_id/request-approval", Group: "approval", Summary: "Request approval for an execution", AuthLevel: "api_key", Tags: []string{"approval", "request"}}, {Method: "GET", Path: "/api/v1/executions/:execution_id/approval-status", Group: "approval", Summary: "Get approval status", AuthLevel: "api_key", Tags: []string{"approval", "status"}}, + {Method: "POST", Path: "/api/v1/executions/:execution_id/approval-response", Group: "approval", Summary: "Resolve a pending approval (approved/rejected/request_changes)", AuthLevel: "api_key", Tags: []string{"approval", "resolve"}}, {Method: "POST", Path: "/api/v1/agents/:node_id/executions/:execution_id/request-approval", Group: "approval", Summary: "Request approval (agent-scoped)", AuthLevel: "api_key", Tags: []string{"approval", "request", "agent-scoped"}}, {Method: "GET", Path: "/api/v1/agents/:node_id/executions/:execution_id/approval-status", Group: "approval", Summary: "Get approval status (agent-scoped)", AuthLevel: "api_key", Tags: []string{"approval", "status", "agent-scoped"}}, {Method: "POST", Path: "/api/v1/webhooks/approval-response", Group: "approval", Summary: "Webhook for approval responses (HMAC-signed)", AuthLevel: "webhook", Tags: []string{"approval", "webhook"}}, From ca5dcc564825d0f8a0b2d363fa3a3a05b1da9796 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Tue, 14 Jul 2026 16:03:09 -0400 Subject: [PATCH 8/9] fix(storage): persist recorded_at on the execution_logs raw INSERT too MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit storeExecutionLogEntryTx — the single writer behind both StoreExecutionLogEntry and StoreExecutionLogEntries — omitted recorded_at from its raw INSERT, so every execution-log row landed NULL and every read silently took the legacy emitted_at fallback: the server-ingestion timestamp was never actually stored. It also stamped entry.RecordedAt only AFTER the insert, mutating the caller's struct with a value that never reached the database. This is the same bug this branch already fixed for workflow_execution_events (the GORM model's autoCreateTime does not apply to raw INSERTs); the sibling path was missed. The default is now stamped before the query is built and recorded_at is bound explicitly, mirroring storeWorkflowExecutionEventTx. The regression test uses distinct emitted_at/recorded_at values so the NULL-read fallback cannot masquerade as persistence, and covers the batch path's zero-value stamping; it fails against the previous INSERT. Co-Authored-By: Claude Fable 5 --- .../coverage_identity_events_reasoner_test.go | 42 +++++++++++++++++++ control-plane/internal/storage/local.go | 16 ++++--- 2 files changed, 53 insertions(+), 5 deletions(-) diff --git a/control-plane/internal/storage/coverage_identity_events_reasoner_test.go b/control-plane/internal/storage/coverage_identity_events_reasoner_test.go index e5174fb0e..349a2addd 100644 --- a/control-plane/internal/storage/coverage_identity_events_reasoner_test.go +++ b/control-plane/internal/storage/coverage_identity_events_reasoner_test.go @@ -188,6 +188,48 @@ func TestExecutionEventAndLogCoverage(t *testing.T) { require.Equal(t, emittedAt, stored[0].RecordedAt.UTC().Truncate(time.Second)) }) + t.Run("persists recorded_at for execution logs", func(t *testing.T) { + ls, ctx := setupLocalStorage(t) + + // Distinct timestamps so the NULL-read fallback (recorded_at -> + // emitted_at) cannot masquerade as persistence: if the raw INSERT + // drops recorded_at again, the listed value comes back as emittedAt + // and this test fails. + emittedAt := time.Now().UTC().Add(-2 * time.Minute).Truncate(time.Second) + recordedAt := time.Now().UTC().Truncate(time.Second) + + entry := &types.ExecutionLogEntry{ + ExecutionID: "exec-logs-recorded", + WorkflowID: "wf-logs-recorded", + Level: "info", + Message: "recorded-at roundtrip", + EmittedAt: emittedAt, + RecordedAt: recordedAt, + } + require.NoError(t, ls.StoreExecutionLogEntry(ctx, entry)) + + stored, err := ls.ListExecutionLogEntries(ctx, "exec-logs-recorded", nil, 10, nil, nil, nil, "") + require.NoError(t, err) + require.Len(t, stored, 1) + require.Equal(t, recordedAt, stored[0].RecordedAt.UTC().Truncate(time.Second)) + + // The batch path funnels through the same Tx helper; a zero + // RecordedAt must be stamped before the INSERT, not after it. + batch := []*types.ExecutionLogEntry{{ + ExecutionID: "exec-logs-recorded-batch", + WorkflowID: "wf-logs-recorded", + Level: "info", + Message: "batch recorded-at", + EmittedAt: emittedAt, + }} + require.NoError(t, ls.StoreExecutionLogEntries(ctx, "exec-logs-recorded-batch", batch)) + stored, err = ls.ListExecutionLogEntries(ctx, "exec-logs-recorded-batch", nil, 10, nil, nil, nil, "") + require.NoError(t, err) + require.Len(t, stored, 1) + require.False(t, stored[0].RecordedAt.IsZero()) + require.NotEqual(t, emittedAt, stored[0].RecordedAt.UTC().Truncate(time.Second)) + }) + t.Run("stores lists and prunes execution logs", func(t *testing.T) { ls, ctx := setupLocalStorage(t) require.EqualError(t, ls.StoreExecutionLogEntry(ctx, nil), "execution log entry is nil") diff --git a/control-plane/internal/storage/local.go b/control-plane/internal/storage/local.go index 99f303b1f..ee8dbb306 100644 --- a/control-plane/internal/storage/local.go +++ b/control-plane/internal/storage/local.go @@ -7881,12 +7881,20 @@ func (ls *LocalStorage) storeExecutionLogEntryTx(ctx context.Context, tx DBTX, e attributes = string(entry.Attributes) } + // Persist recorded_at explicitly: the GORM model's autoCreateTime does + // not apply to this raw INSERT, so omitting the column stores NULL and + // every read silently takes the legacy emitted_at fallback. + if entry.RecordedAt.IsZero() { + entry.RecordedAt = time.Now().UTC() + } + query := ` INSERT INTO execution_logs ( execution_id, workflow_id, run_id, root_workflow_id, parent_execution_id, sequence, agent_node_id, reasoner_id, level, source, event_type, message, attributes, - system_generated, sdk_language, attempt, span_id, step_id, error_category, emitted_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)` + system_generated, sdk_language, attempt, span_id, step_id, error_category, emitted_at, + recorded_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)` result, err := tx.ExecContext(ctx, query, entry.ExecutionID, @@ -7909,6 +7917,7 @@ func (ls *LocalStorage) storeExecutionLogEntryTx(ctx context.Context, tx DBTX, e entry.StepID, entry.ErrorCategory, entry.EmittedAt, + entry.RecordedAt, ) if err != nil { return fmt.Errorf("failed to insert execution log entry: %w", err) @@ -7916,9 +7925,6 @@ func (ls *LocalStorage) storeExecutionLogEntryTx(ctx context.Context, tx DBTX, e if id, err := result.LastInsertId(); err == nil { entry.EventID = id } - if entry.RecordedAt.IsZero() { - entry.RecordedAt = time.Now().UTC() - } return nil } From 9c4e2ae417e80e736630fbd19e5f7077f72bc163 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Thu, 16 Jul 2026 21:14:43 -0400 Subject: [PATCH 9/9] test(storage): align clinch execution-log test with persisted recorded_at The recorded_at-persistence fix (raw INSERT no longer leaves the column NULL) made the legacy assertion here fail: it expected the NULL-read fallback (recorded_at -> emitted_at) for a freshly stored entry. Give the entry an explicit RecordedAt and assert it round-trips; the NULL fallback branch is still covered by the forced-NULL UPDATE above. Latent in the branch before the main merge - conflict state had kept full CI from running on the previous head. Co-Authored-By: Claude Fable 5 --- .../internal/storage/coverage_storage_clinch_test.go | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/control-plane/internal/storage/coverage_storage_clinch_test.go b/control-plane/internal/storage/coverage_storage_clinch_test.go index cacc87988..484865cd2 100644 --- a/control-plane/internal/storage/coverage_storage_clinch_test.go +++ b/control-plane/internal/storage/coverage_storage_clinch_test.go @@ -249,6 +249,7 @@ func TestStorageClinchConfigAndExecutionLogBranches(t *testing.T) { Message: "second error", Attributes: json.RawMessage(`{"trace":"abc"}`), EmittedAt: now.Add(-2 * time.Minute), + RecordedAt: now.Add(-90 * time.Second), } entryThree := &types.ExecutionLogEntry{ ExecutionID: "exec-log-branches", @@ -277,7 +278,9 @@ func TestStorageClinchConfigAndExecutionLogBranches(t *testing.T) { require.NoError(t, err) require.Len(t, filtered, 1) require.Equal(t, entryTwo.EventID, filtered[0].EventID) - require.Equal(t, entryTwo.EmittedAt, filtered[0].RecordedAt) + // recorded_at is persisted by the INSERT now; an explicit value must + // round-trip instead of falling back to emitted_at. + require.Equal(t, entryTwo.RecordedAt, filtered[0].RecordedAt.UTC().Truncate(time.Second)) require.JSONEq(t, `{"trace":"abc"}`, string(filtered[0].Attributes)) require.NoError(t, ls.PruneExecutionLogEntries(ctx, "exec-log-branches", 2, now.Add(-150*time.Second)))