feat(control-plane): agent-mode steering verbs, events query resource, and --json lifecycle output

control-plane/internal/cli/agent_commands.go

@@ -70,6 +70,7 @@ func NewAgentCommand() *cobra.Command {
 	cmd.AddCommand(newAgentSearchCmd())
 	cmd.AddCommand(newAgentQueryCmd())
 	cmd.AddCommand(newAgentRunCmd())
+	cmd.AddCommand(newAgentExecCmd())
 	cmd.AddCommand(newAgentAgentSummaryCmd())
 	cmd.AddCommand(newAgentKBCmd())
 	cmd.AddCommand(newAgentBatchCmd())
@@ -169,6 +170,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
@@ -178,10 +180,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{}
@@ -194,6 +205,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
 			}
@@ -225,10 +239,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")
@@ -587,6 +602,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)
 			}
 		}
 
@@ -633,6 +650,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:
@@ -707,20 +762,22 @@ 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"},
 					{"name": "offset", "short": "", "type": "int"},
 					{"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",
@@ -736,6 +793,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 <execution_id> [--reason <text>]",
+				"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 <execution_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 <execution_id> [--reason <text>]",
+				"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 <execution_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 <execution_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 <execution_id> --decision <approved|rejected|request_changes> [--reason <text>]",
+				"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",
@@ -797,7 +914,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", "GET /api/v1/agentic/reasoners", "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", "GET /api/v1/agentic/reasoners", "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{}{

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
+}