diff --git a/docs/openapi.json b/docs/openapi.json index 123c9858..f8089a94 100644 --- a/docs/openapi.json +++ b/docs/openapi.json @@ -4148,6 +4148,212 @@ "x-cyberagent-read-only": true } }, + "/api/v1/runs/{run_id}/fanout-executions": { + "get": { + "operationId": "listRunFanoutExecutions", + "summary": "List one Fan-out plan's execution history", + "description": "Returns the bounded shard execution history of one read-only Fan-out plan; shard rows carry only status, bounded counts, and model identity.", + "tags": [ + "Analysis" + ], + "parameters": [ + { + "name": "run_id", + "in": "path", + "description": "Run identity", + "required": true, + "schema": { + "maxLength": 256, + "minLength": 1, + "pattern": "^[^/\\\\\\x00-\\x1f\\x7f]+$", + "type": "string" + } + }, + { + "name": "plan_id", + "in": "query", + "description": "Exact Fan-out plan identity", + "schema": { + "maxLength": 256, + "minLength": 1, + "pattern": "^[^/\\\\\\x00-\\x1f\\x7f]+$", + "type": "string" + } + } + ], + "responses": { + "200": { + "description": "Successful read", + "content": { + "application/json": { + "schema": { + "additionalProperties": false, + "properties": { + "data": { + "$ref": "#/components/schemas/FanoutExecutionsListView" + }, + "request_id": { + "maxLength": 256, + "minLength": 1, + "type": "string" + }, + "version": { + "const": "api.v1", + "type": "string" + } + }, + "required": [ + "version", + "request_id", + "data" + ], + "type": "object" + } + } + } + }, + "400": { + "$ref": "#/components/responses/BadRequest" + }, + "401": { + "$ref": "#/components/responses/Unauthorized" + }, + "403": { + "$ref": "#/components/responses/Forbidden" + }, + "404": { + "$ref": "#/components/responses/NotFound" + }, + "414": { + "$ref": "#/components/responses/RequestTooLarge" + }, + "429": { + "$ref": "#/components/responses/ResourceExhausted" + }, + "500": { + "$ref": "#/components/responses/InternalError" + } + }, + "x-cyberagent-read-only": true + } + }, + "/api/v1/runs/{run_id}/fanout-executions/{execution_id}/cancel": { + "post": { + "operationId": "cancelRunFanoutExecution", + "summary": "Cancel one read-only Fan-out execution", + "description": "Cancels one non-terminal read-only Fan-out execution under a fresh Run execution lease fence; terminal executions return their stored state unchanged.", + "tags": [ + "Control" + ], + "parameters": [ + { + "name": "run_id", + "in": "path", + "description": "Run identity", + "required": true, + "schema": { + "maxLength": 256, + "minLength": 1, + "pattern": "^[^/\\\\\\x00-\\x1f\\x7f]+$", + "type": "string" + } + }, + { + "name": "execution_id", + "in": "path", + "description": "Read-only Fan-out execution identity", + "required": true, + "schema": { + "maxLength": 256, + "minLength": 1, + "pattern": "^[^/\\\\\\x00-\\x1f\\x7f]+$", + "type": "string" + } + } + ], + "responses": { + "200": { + "description": "Control request accepted or idempotently replayed", + "content": { + "application/json": { + "schema": { + "additionalProperties": false, + "properties": { + "data": { + "$ref": "#/components/schemas/FanoutExecutionView" + }, + "request_id": { + "maxLength": 256, + "minLength": 1, + "type": "string" + }, + "version": { + "const": "api.v1", + "type": "string" + } + }, + "required": [ + "version", + "request_id", + "data" + ], + "type": "object" + } + } + } + }, + "400": { + "$ref": "#/components/responses/BadRequest" + }, + "401": { + "$ref": "#/components/responses/Unauthorized" + }, + "403": { + "$ref": "#/components/responses/Forbidden" + }, + "404": { + "$ref": "#/components/responses/NotFound" + }, + "409": { + "$ref": "#/components/responses/Conflict" + }, + "412": { + "$ref": "#/components/responses/FailedPrecondition" + }, + "413": { + "$ref": "#/components/responses/RequestEntityTooLarge" + }, + "414": { + "$ref": "#/components/responses/RequestTooLarge" + }, + "415": { + "$ref": "#/components/responses/UnsupportedMediaType" + }, + "429": { + "$ref": "#/components/responses/ResourceExhausted" + }, + "500": { + "$ref": "#/components/responses/InternalError" + } + }, + "security": [ + { + "ControlBearerAuth": [] + } + ], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/FanoutExecutionCancelRequestView" + } + } + } + }, + "x-cyberagent-read-only": false + } + }, "/api/v1/runs/{run_id}/fanout-plans": { "get": { "operationId": "listRunFanoutPlans", @@ -13659,6 +13865,22 @@ ], "type": "object" }, + "FanoutExecutionCancelRequestView": { + "additionalProperties": false, + "properties": { + "confirm_cancel": { + "type": "boolean" + }, + "version": { + "type": "string" + } + }, + "required": [ + "version", + "confirm_cancel" + ], + "type": "object" + }, "FanoutExecutionShardView": { "additionalProperties": false, "properties": { @@ -13790,6 +14012,31 @@ ], "type": "object" }, + "FanoutExecutionsListView": { + "additionalProperties": false, + "properties": { + "items": { + "items": { + "$ref": "#/components/schemas/FanoutExecutionView" + }, + "type": "array" + }, + "plan_id": { + "maxLength": 256, + "minLength": 1, + "type": "string" + }, + "protocol_version": { + "type": "string" + } + }, + "required": [ + "protocol_version", + "plan_id", + "items" + ], + "type": "object" + }, "FanoutPlanView": { "additionalProperties": false, "properties": { diff --git a/internal/app/api_command.go b/internal/app/api_command.go index 5ace5233..6d8481f7 100644 --- a/internal/app/api_command.go +++ b/internal/app/api_command.go @@ -231,6 +231,8 @@ func (a *App) apiServeCommand(ctx context.Context, args []string) error { HostCommandProposalController: hostCommandProposalControl, ModelControlController: modelControl, PriceSnapshotController: a.store, + FanoutExecutionController: application.NewReadOnlyFanoutExecutionService( + a.store, a.router, a.checker), ProviderCredentialController: providerCredentialControl, FileEditReviewController: fileEditReview, FileEditProposalController: fileEditProposal, diff --git a/internal/application/readonly_fanout_execution.go b/internal/application/readonly_fanout_execution.go index 1c9e99a3..dc48e26f 100644 --- a/internal/application/readonly_fanout_execution.go +++ b/internal/application/readonly_fanout_execution.go @@ -39,6 +39,8 @@ type ReadOnlyFanoutExecutionStore interface { keyDigest string) (domain.ReadOnlyFanoutExecutionOperation, bool, error) GetReadOnlyFanoutExecution(ctx context.Context, id string) (domain.ReadOnlyFanoutExecution, error) + ListReadOnlyFanoutExecutions(ctx context.Context, + planID string, limit int) ([]domain.ReadOnlyFanoutExecution, error) CreateReadOnlyFanoutExecution(ctx context.Context, lease domain.RunExecutionLease, execution domain.ReadOnlyFanoutExecution, operation domain.ReadOnlyFanoutExecutionOperation, @@ -886,3 +888,76 @@ func normalizeReadOnlyFanoutExecutionDecision(decision policy.Decision) policy.D } return decision } +// ListReadOnlyFanoutExecutions projects one plan's execution history. +func (s *ReadOnlyFanoutExecutionService) ListReadOnlyFanoutExecutions(ctx context.Context, + planID string, limit int, +) ([]domain.ReadOnlyFanoutExecution, error) { + if s == nil || s.store == nil { + return nil, apperror.New(apperror.CodeFailedPrecondition, + "read-only fan-out execution dependencies are required") + } + executions, err := s.store.ListReadOnlyFanoutExecutions(ctx, planID, limit) + return executions, apperror.Normalize(err) +} + +// CancelReadOnlyFanoutExecution cancels one non-terminal fan-out execution +// under a fresh lease fence: the remainder of pending shards is cancelled and +// the execution finalizes cancelled. Terminal executions return their stored +// state unchanged, so the control endpoint stays idempotent. +func (s *ReadOnlyFanoutExecutionService) CancelReadOnlyFanoutExecution(ctx context.Context, + executionID, requestedBy, operationKey string, +) (domain.ReadOnlyFanoutExecution, error) { + if s == nil || s.store == nil { + return domain.ReadOnlyFanoutExecution{}, apperror.New( + apperror.CodeFailedPrecondition, + "read-only fan-out execution dependencies are required") + } + executionID = strings.TrimSpace(executionID) + if !domain.ValidAgentID(executionID) || strings.TrimSpace(operationKey) == "" || + strings.TrimSpace(requestedBy) == "" { + return domain.ReadOnlyFanoutExecution{}, apperror.New( + apperror.CodeInvalidArgument, + "read-only fan-out cancellation request is invalid") + } + execution, err := s.store.GetReadOnlyFanoutExecution(ctx, executionID) + if err != nil { + return domain.ReadOnlyFanoutExecution{}, apperror.Normalize(err) + } + if execution.Status.Terminal() { + return execution, nil + } + plan, err := s.store.GetReadOnlyFanoutPlan(ctx, execution.PlanID) + if err != nil { + return domain.ReadOnlyFanoutExecution{}, apperror.Normalize(err) + } + run, _, linkedSession, workspace, err := s.loadExecutionBinding(ctx, plan) + if err != nil { + return domain.ReadOnlyFanoutExecution{}, err + } + result := execution + err = withRunExecutionLease(ctx, s.store, run.ID, s.leaseOwner, s.leasePolicy, + func(leaseCtx context.Context, lease domain.RunExecutionLease) error { + currentRun, _, currentSession, currentWorkspace, err := + s.loadExecutionBinding(leaseCtx, plan) + if err != nil { + return err + } + if currentSession.ID != linkedSession.ID || currentWorkspace.ID != workspace.ID || + currentRun.Config.ModelRoute != run.Config.ModelRoute { + return apperror.New(apperror.CodeConflict, + "read-only fan-out Run binding changed before cancellation") + } + if _, err := s.store.CancelReadOnlyFanoutExecutionRemainder(leaseCtx, lease, + executionID, "cancelled", "operator_cancelled"); err != nil { + return err + } + final, _, err := s.store.FinalizeReadOnlyFanoutExecution(leaseCtx, lease, + executionID, domain.ReadOnlyFanoutExecutionCancelled, "operator_cancelled") + if err != nil { + return err + } + result = final + return nil + }) + return result, apperror.Normalize(err) +} diff --git a/internal/desktop/control_plane.go b/internal/desktop/control_plane.go index 0b809564..c9bba3c1 100644 --- a/internal/desktop/control_plane.go +++ b/internal/desktop/control_plane.go @@ -302,6 +302,8 @@ func OpenControlPlane(config ControlPlaneConfig) (*ControlPlane, error) { HostCommandProposalController: hostCommandProposals, ModelControlController: modelControl, PriceSnapshotController: stateStore, + FanoutExecutionController: application.NewReadOnlyFanoutExecutionService( + stateStore, models.Router(), checker), ProviderCredentialController: providerCredentialControl, FileEditReviewController: fileEditReview, FileEditProposalController: fileEditProposal, diff --git a/internal/httpapi/fanout_execution_control.go b/internal/httpapi/fanout_execution_control.go new file mode 100644 index 00000000..b3ee08ee --- /dev/null +++ b/internal/httpapi/fanout_execution_control.go @@ -0,0 +1,180 @@ +package httpapi + +import ( + "context" + "net/http" + "strconv" + "strings" + + "cyberagent-workbench/internal/apperror" + "cyberagent-workbench/internal/domain" +) + +const ( + FanoutExecutionsPathTemplate = "/api/v1/runs/{run_id}/fanout-executions" + FanoutExecutionCancelPathTemplate = "/api/v1/runs/{run_id}/fanout-executions/{execution_id}/cancel" + FanoutExecutionsListProtocolVersion = "readonly_fanout_executions.v1" +) + +// FanoutExecutionController is the bounded read/list + cancel surface for +// read-only fan-out executions. +type FanoutExecutionController interface { + ListReadOnlyFanoutExecutions(context.Context, string, int) ([]domain.ReadOnlyFanoutExecution, error) + CancelReadOnlyFanoutExecution(context.Context, string, string, string) (domain.ReadOnlyFanoutExecution, error) +} + +// FanoutExecutionCancelRequestView is the explicit-confirmation control body. +type FanoutExecutionCancelRequestView struct { + Version string `json:"version"` + ConfirmCancel bool `json:"confirm_cancel"` +} + +// FanoutExecutionsListView projects one plan's execution history. +type FanoutExecutionsListView struct { + ProtocolVersion string `json:"protocol_version"` + PlanID string `json:"plan_id"` + Items []FanoutExecutionView `json:"items"` +} + +func (a *API) listFanoutExecutions(request *http.Request, runID string) (any, *Page, error) { + if err := validateSingleQueryValues(request.URL.Query(), "plan_id", "limit"); err != nil { + return nil, nil, err + } + planID := strings.TrimSpace(request.URL.Query().Get("plan_id")) + if planID == "" || !domain.ValidAgentID(planID) { + return nil, nil, apperror.New(apperror.CodeInvalidArgument, + "fan-out execution list requires a valid plan_id") + } + limit := 50 + if raw := request.URL.Query().Get("limit"); raw != "" { + parsed, err := strconv.Atoi(raw) + if err != nil || parsed < 1 || parsed > 100 { + return nil, nil, apperror.New(apperror.CodeInvalidArgument, + "fan-out execution limit must be between 1 and 100") + } + limit = parsed + } + if _, err := a.store.GetRun(request.Context(), runID); err != nil { + return nil, nil, err + } + executions, err := a.fanoutExecutionController.ListReadOnlyFanoutExecutions( + request.Context(), planID, limit) + if err != nil { + return nil, nil, err + } + view := FanoutExecutionsListView{ProtocolVersion: FanoutExecutionsListProtocolVersion, + PlanID: planID, Items: make([]FanoutExecutionView, 0, len(executions))} + for _, execution := range executions { + view.Items = append(view.Items, fanoutExecutionFullView(execution)) + } + return view, nil, nil +} + +func matchFanoutExecutionCancelPath(requestPath string) (string, string, bool) { + const prefix = "/api/v1/runs/" + if !strings.HasPrefix(requestPath, prefix) { + return "", "", false + } + segments := strings.Split(strings.TrimPrefix(requestPath, prefix), "/") + if len(segments) != 4 || segments[0] == "" || segments[1] != "fanout-executions" || + segments[2] == "" || segments[3] != "cancel" { + return "", "", false + } + return segments[0], segments[2], true +} + +func (a *API) serveFanoutExecutionCancel(writer http.ResponseWriter, request *http.Request, + requestID, runID, executionID string, +) { + if a.fanoutExecutionController == nil { + a.writeError(writer, requestID, + apperror.New(apperror.CodeNotFound, "HTTP API endpoint was not found"), + http.StatusNotFound) + return + } + if !a.controlEnabled { + a.writeError(writer, requestID, + apperror.New(apperror.CodeNotFound, "HTTP API endpoint was not found"), + http.StatusNotFound) + return + } + if !a.authorized(request, a.controlTokenHash) { + writer.Header().Set("WWW-Authenticate", `Bearer realm="CyberAgent Control API"`) + a.writeError(writer, requestID, + apperror.New(apperror.CodePolicyDenied, "valid control bearer authorization is required"), + http.StatusUnauthorized) + return + } + if request.Method != http.MethodPost { + writer.Header().Set("Allow", http.MethodPost) + a.writeError(writer, requestID, + apperror.New(apperror.CodeInvalidArgument, + "fan-out execution cancellation only supports POST"), + http.StatusMethodNotAllowed) + return + } + if err := validatePathIdentity(runID); err != nil { + a.writeError(writer, requestID, err, 0) + return + } + if err := validatePathIdentity(executionID); err != nil { + a.writeError(writer, requestID, err, 0) + return + } + if err := rejectQuery(request.URL.Query()); err != nil { + a.writeError(writer, requestID, err, 0) + return + } + var body struct { + Version string `json:"version"` + ConfirmCancel bool `json:"confirm_cancel"` + } + strictBody, err := readStrictControlBody(request, "fan-out execution cancellation") + if err != nil { + a.writeError(writer, requestID, err, runOperationErrorStatus(err)) + return + } + if err := decodeStrictRunOperation(strictBody, &body, "fan-out execution cancellation"); err != nil { + a.writeError(writer, requestID, err, 0) + return + } + if body.Version != "readonly_fanout_cancel.v1" || !body.ConfirmCancel { + a.writeError(writer, requestID, apperror.New(apperror.CodeInvalidArgument, + "fan-out execution cancellation requires explicit confirmation"), + http.StatusBadRequest) + return + } + execution, err := a.fanoutExecutionController.CancelReadOnlyFanoutExecution( + request.Context(), executionID, "web_operator", requestID) + if err != nil { + a.writeError(writer, requestID, err, 0) + return + } + if execution.RunID != runID { + a.writeError(writer, requestID, apperror.New(apperror.CodeNotFound, + "fan-out execution is not part of the run"), http.StatusNotFound) + return + } + a.writeSuccessStatus(writer, requestID, fanoutExecutionFullView(execution), nil, http.StatusOK) +} + +func fanoutExecutionFullView(value domain.ReadOnlyFanoutExecution) FanoutExecutionView { + view := FanoutExecutionView{ID: value.ID, Status: string(value.Status), + Parallelism: value.Parallelism, MaxOutputTokensPerShard: value.MaxOutputTokensPerShard, + RequestedBy: value.RequestedBy, StopCode: value.StopCode, + StartedAt: value.StartedAt, UpdatedAt: value.UpdatedAt, + FinishedAt: value.FinishedAt, Shards: make([]FanoutExecutionShardView, 0, len(value.Shards))} + for _, shard := range value.Shards { + view.Shards = append(view.Shards, FanoutExecutionShardView{ + Ordinal: shard.Ordinal, Status: string(shard.Status), + AttemptCount: shard.AttemptCount, CurrentAttempt: shard.CurrentAttempt, + Provider: shard.Provider, Model: shard.Model, + InputTokens: shard.InputTokens, OutputTokens: shard.OutputTokens, + TotalTokens: shard.TotalTokens, ElapsedMillis: shard.ElapsedMillis, + FindingCount: shard.FindingCount, ErrorCode: shard.ErrorCode, + StartedAt: shard.StartedAt, FinishedAt: shard.FinishedAt, + }) + } + return view +} + diff --git a/internal/httpapi/handlers.go b/internal/httpapi/handlers.go index 0df53db9..ebcb1660 100644 --- a/internal/httpapi/handlers.go +++ b/internal/httpapi/handlers.go @@ -439,6 +439,8 @@ func (a *API) routeRuns(request *http.Request, segments []string) (any, *Page, e return a.runDelegations(request, segments[1]) case "fanout-plans": return a.runFanoutPlans(request, segments[1]) + case "fanout-executions": + return a.listFanoutExecutions(request, segments[1]) case "reports": return a.runFindingReports(request, segments[1]) case "events": diff --git a/internal/httpapi/openapi.go b/internal/httpapi/openapi.go index 6f51c449..c014ae3a 100644 --- a/internal/httpapi/openapi.go +++ b/internal/httpapi/openapi.go @@ -221,6 +221,7 @@ func GenerateOpenAPI() ([]byte, error) { func openAPIOperationSpecs() []openAPIOperationSpec { runID := pathIdentityParameter("run_id", "Run identity") + executionID := pathIdentityParameter("execution_id", "Read-only Fan-out execution identity") verificationPlanID := pathIdentityParameter("plan_id", "Verification plan identity") verificationPlanItemOrdinal := openAPIParameter{Name: "ordinal", In: "path", Description: "Verification plan item ordinal", Required: true, @@ -632,6 +633,11 @@ func openAPIOperationSpecs() []openAPIOperationSpec { Description: "Returns bounded plan metadata and the latest shard execution summary without file manifests or model report JSON.", DataType: reflect.TypeOf(FanoutPlanView{}), Collection: true, Paginated: true, NotFound: true, Parameters: append([]openAPIParameter{runID}, paginationParameters()...)}, + {Path: FanoutExecutionsPathTemplate, OperationID: "listRunFanoutExecutions", + Summary: "List one Fan-out plan's execution history", Tag: "Analysis", + Description: "Returns the bounded shard execution history of one read-only Fan-out plan; shard rows carry only status, bounded counts, and model identity.", + DataType: reflect.TypeOf(FanoutExecutionsListView{}), NotFound: true, + Parameters: []openAPIParameter{runID, identityQueryParameter("plan_id", "Exact Fan-out plan identity")}}, {Path: "/api/v1/runs/{run_id}/reports", OperationID: "listRunFindingReports", Summary: "List Finding Report summaries", Tag: "Reports", Description: "Returns bounded report counts and severity summaries without Finding narratives.", @@ -677,6 +683,14 @@ func openAPIOperationSpecs() []openAPIOperationSpec { Required: true, Schema: map[string]any{"type": "string", "minLength": domain.MinModelCancellationKeyBytes, "maxLength": domain.MaxModelCancellationKeyBytes, "pattern": `^\S+$`}}, }}, + {Path: FanoutExecutionCancelPathTemplate, Method: http.MethodPost, + OperationID: "cancelRunFanoutExecution", + Summary: "Cancel one read-only Fan-out execution", Tag: "Control", + Description: "Cancels one non-terminal read-only Fan-out execution under a fresh Run execution lease fence; terminal executions return their stored state unchanged.", + DataType: reflect.TypeOf(FanoutExecutionView{}), + RequestType: reflect.TypeOf(FanoutExecutionCancelRequestView{}), + Control: true, NotFound: true, + SuccessStatus: http.StatusOK, Parameters: []openAPIParameter{runID, executionID}}, {Path: SpecialistModelCancellationPathTemplate, Method: http.MethodPost, OperationID: "requestSpecialistModelCancellation", Summary: "Cancel an active Specialist model call", Tag: "Control", diff --git a/internal/httpapi/openapi_test.go b/internal/httpapi/openapi_test.go index 6e552a58..3a5a84e6 100644 --- a/internal/httpapi/openapi_test.go +++ b/internal/httpapi/openapi_test.go @@ -139,6 +139,8 @@ func TestOpenAPIDocumentIsDeterministicCapabilitySeparatedAndSecretFree(t *testi item.Post.OperationID == "qualifyModelHarness") || (path == PriceSnapshotsPath && item.Post.OperationID == "importPriceSnapshot") || + (path == FanoutExecutionCancelPathTemplate && + item.Post.OperationID == "cancelRunFanoutExecution") || (path == ProviderCredentialPathTemplate && item.Post.OperationID == "changeProviderCredential") || (path == FileEditProposalPathTemplate && @@ -239,6 +241,7 @@ func TestOpenAPIDocumentIsDeterministicCapabilitySeparatedAndSecretFree(t *testi path == ModelRouteControlPathTemplate || path == ProviderDiagnosticPath || path == ModelHarnessQualificationPath || path == PriceSnapshotsPath || + path == FanoutExecutionCancelPathTemplate || path == ProviderCredentialPathTemplate || path == FileEditProposalPathTemplate || path == FileEditReviewPathTemplate || path == FileEditApplyPathTemplate || @@ -598,6 +601,8 @@ func TestOpenAPIRoutesMatchAuthenticatedLiveHandlers(t *testing.T) { fixture.api.modelControlController = application.NewModelControlService( fixture.api.modelRegistry, fixture.store) fixture.api.priceSnapshotController = fixture.store + fixture.api.fanoutExecutionController = application.NewReadOnlyFanoutExecutionService( + fixture.store, llm.NewDefaultRouter(), policy.NewDefaultChecker()) credentialStore := credential.NewMemoryStore() fixture.api.providerCredentialController = application.NewProviderCredentialService( credentialStore) @@ -696,6 +701,7 @@ func TestOpenAPIRoutesMatchAuthenticatedLiveHandlers(t *testing.T) { "{note_id}": fixture.notes[0].ID, "{artifact_id}": fixture.artifactID, "{report_id}": "report-openapi-missing-0001", + "{execution_id}": "fanout-execution-openapi-missing-0001", "{approval_id}": approvalRecord.ID, "{proposal_id}": "controlled-command-proposal-openapi", "{edit_id}": fileEditRecord.ID, @@ -756,12 +762,16 @@ func TestOpenAPIRoutesMatchAuthenticatedLiveHandlers(t *testing.T) { requestPath += "?format=json" } else if spec.OperationID == "getDockerSandboxStatus" { requestPath += "?admission_id=docker-sandbox-admission-openapi" + } else if spec.OperationID == "listRunFanoutExecutions" { + requestPath += "?plan_id=plan-openapi-fanout" } t.Run(spec.OperationID, func(t *testing.T) { var response *httptest.ResponseRecorder expectedStatus := http.StatusOK if spec.OperationID == "getRunFindingReport" { expectedStatus = http.StatusNotFound + } else if spec.OperationID == "cancelRunFanoutExecution" { + expectedStatus = http.StatusNotFound } else if spec.OperationID == "getWorkspaceRepositoryCommitFilePreview" { expectedStatus = http.StatusPreconditionFailed } @@ -828,6 +838,8 @@ func TestOpenAPIRoutesMatchAuthenticatedLiveHandlers(t *testing.T) { body = `{"version":"provider_diagnostic.v1","provider":"mock","model":"mock-code","confirm_diagnostic":true}` } else if spec.Path == ModelHarnessQualificationPath { body = `{"version":"model_harness_qualification.v1","provider":"mock","model":"mock-code","confirm_qualification":true}` + } else if spec.Path == FanoutExecutionCancelPathTemplate { + body = `{"version":"readonly_fanout_cancel.v1","confirm_cancel":true}` } else if spec.Path == PriceSnapshotsPath { encoded, marshalErr := json.Marshal(PriceSnapshotImportRequestView{ Version: pricing.ProtocolVersion, @@ -929,9 +941,11 @@ func TestOpenAPIRoutesMatchAuthenticatedLiveHandlers(t *testing.T) { if statusErr != nil { t.Fatal(statusErr) } - expectedStatus, statusErr = strconv.Atoi(status) - if statusErr != nil { - t.Fatal(statusErr) + if spec.OperationID != "cancelRunFanoutExecution" { + expectedStatus, statusErr = strconv.Atoi(status) + if statusErr != nil { + t.Fatal(statusErr) + } } } else { response = fixture.get(t, requestPath) diff --git a/internal/httpapi/server.go b/internal/httpapi/server.go index b55a9be1..cfa58bfa 100644 --- a/internal/httpapi/server.go +++ b/internal/httpapi/server.go @@ -255,6 +255,7 @@ type Config struct { HostCommandProposalController HostCommandProposalController ModelControlController ModelControlController PriceSnapshotController PriceSnapshotController + FanoutExecutionController FanoutExecutionController ProviderCredentialController ProviderCredentialController FileEditReviewController FileEditReviewController FileEditProposalController FileEditProposalController @@ -312,6 +313,7 @@ type API struct { hostCommandProposalController HostCommandProposalController modelControlController ModelControlController priceSnapshotController PriceSnapshotController + fanoutExecutionController FanoutExecutionController providerCredentialController ProviderCredentialController fileEditReviewController FileEditReviewController fileEditProposalController FileEditProposalController @@ -546,7 +548,8 @@ func New(store Store, config Config) (*API, error) { controlledCommandProposalController: config.ControlledCommandProposalController, hostCommandProposalController: config.HostCommandProposalController, modelControlController: config.ModelControlController, - priceSnapshotController: config.PriceSnapshotController, + priceSnapshotController: config.PriceSnapshotController, + fanoutExecutionController: config.FanoutExecutionController, providerCredentialController: config.ProviderCredentialController, fileEditReviewController: config.FileEditReviewController, fileEditProposalController: config.FileEditProposalController, @@ -822,6 +825,10 @@ func (a *API) ServeHTTP(writer http.ResponseWriter, request *http.Request) { a.serveModelCancellation(tracked, request, requestID, runID) return } + if runID, executionID, matched := matchFanoutExecutionCancelPath(request.URL.Path); matched { + a.serveFanoutExecutionCancel(tracked, request, requestID, runID, executionID) + return + } if !a.authorized(request, a.tokenHash) { tracked.Header().Set("WWW-Authenticate", `Bearer realm="CyberAgent API"`) a.writeError(tracked, requestID, diff --git a/web/package-lock.json b/web/package-lock.json index 9096da06..8d8ac2a8 100644 --- a/web/package-lock.json +++ b/web/package-lock.json @@ -488,9 +488,6 @@ "arm64" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MIT", "optional": true, "os": [ @@ -508,9 +505,6 @@ "arm64" ], "dev": true, - "libc": [ - "musl" - ], "license": "MIT", "optional": true, "os": [ @@ -528,9 +522,6 @@ "ppc64" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MIT", "optional": true, "os": [ @@ -548,9 +539,6 @@ "s390x" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MIT", "optional": true, "os": [ @@ -568,9 +556,6 @@ "x64" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MIT", "optional": true, "os": [ @@ -588,9 +573,6 @@ "x64" ], "dev": true, - "libc": [ - "musl" - ], "license": "MIT", "optional": true, "os": [ @@ -1901,9 +1883,6 @@ "arm64" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MPL-2.0", "optional": true, "os": [ @@ -1925,9 +1904,6 @@ "arm64" ], "dev": true, - "libc": [ - "musl" - ], "license": "MPL-2.0", "optional": true, "os": [ @@ -1949,9 +1925,6 @@ "x64" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MPL-2.0", "optional": true, "os": [ @@ -1973,9 +1946,6 @@ "x64" ], "dev": true, - "libc": [ - "musl" - ], "license": "MPL-2.0", "optional": true, "os": [ diff --git a/web/src/api/client.ts b/web/src/api/client.ts index 25b228a4..e4c3be43 100644 --- a/web/src/api/client.ts +++ b/web/src/api/client.ts @@ -1,6 +1,9 @@ import { consumeSSE } from "./sse"; import type { ApprovalDecisionControlRequestView, + FanoutExecutionCancelRequestView, + FanoutExecutionsListView, + FanoutExecutionView, ApprovalDecisionControlView, ApprovalQueueView, ArtifactView, @@ -4162,6 +4165,34 @@ export class CyberAgentClient { return parseSpecialistModelCancellation(result, runID, agentID, body); } + async getRunFanoutExecutions(runID: string, planID: string, + signal?: AbortSignal): Promise { + if (!boundedIdentity(runID) || !boundedIdentity(planID)) { + throw new Error("Normalized Run and Fan-out plan identities are required"); + } + const result = await this.get( + `/runs/${encodeURIComponent(runID)}/fanout-executions?plan_id=${encodeURIComponent(planID)}`, + {}, signal, + ); + return parseFanoutExecutions(result, runID, planID); + } + + async cancelRunFanoutExecution(runID: string, executionID: string, + body: FanoutExecutionCancelRequestView, idempotencyKey: string, + signal?: AbortSignal): Promise { + if (!this.hasControl) { + throw new Error("A control bearer token is required for this operation"); + } + if (!boundedIdentity(runID) || !boundedIdentity(executionID) || + body.version !== "readonly_fanout_cancel.v1" || body.confirm_cancel !== true) { + throw new Error("Normalized Run/execution identities and explicit confirmation are required"); + } + const result = await this.sendControl( + `/runs/${encodeURIComponent(runID)}/fanout-executions/${encodeURIComponent(executionID)}/cancel`, + body, idempotencyKey, signal, + ); + return parseFanoutExecution(result); + } async selectPlanDirection(runID: string, body: PlanDirectionControlRequestView, idempotencyKey: string, signal?: AbortSignal): Promise { if (!this.hasPlanDelivery) { @@ -4495,3 +4526,40 @@ export class CyberAgentClient { return `${url.pathname}${url.search}`; } } +function parseFanoutExecutions(value: unknown, runID: string, + planID: string): FanoutExecutionsListView { + if (!isRecord(value) || + value.protocol_version !== "readonly_fanout_executions.v1" || + value.plan_id !== planID || !Array.isArray(value.items) || value.items.length > 100) { + throw new APIRequestError("Fan-out execution list is invalid", "INVALID_RESPONSE", 502); + } + return { protocol_version: value.protocol_version, plan_id: value.plan_id, + items: value.items.map(parseFanoutExecution) } as FanoutExecutionsListView; +} + +function parseFanoutExecution(value: unknown): FanoutExecutionView { + if (!isRecord(value) || !boundedIdentity(String(value.id)) || + !["running", "completed", "failed", "cancelled"].includes(String(value.status)) || + !safeBoundedCount(value.parallelism, 6) || !safeBoundedCount(value.max_output_tokens_per_shard, 4096) || + !boundedIdentity(String(value.requested_by)) || !validDate(value.started_at) || + !validDate(value.updated_at) || !Array.isArray(value.shards) || value.shards.length > 6) { + throw new APIRequestError("Fan-out execution is invalid", "INVALID_RESPONSE", 502); + } + if (value.finished_at !== undefined && !validDate(value.finished_at)) { + throw new APIRequestError("Fan-out execution finished_at is invalid", "INVALID_RESPONSE", 502); + } + const shards = value.shards.map((shard) => { + if (!isRecord(shard) || !safeBoundedCount(shard.ordinal, 6) || + !["pending", "running", "completed", "failed", "cancelled"].includes(String(shard.status)) || + !safeBoundedCount(shard.attempt_count, 64) || !safeBoundedCount(shard.current_attempt, 64) || + !safeBoundedCount(shard.input_tokens, 1_000_000_000) || + !safeBoundedCount(shard.output_tokens, 1_000_000_000) || + !safeBoundedCount(shard.total_tokens, 1_000_000_000) || + !safeBoundedCount(shard.elapsed_millis, 1_000_000_000) || + !safeBoundedCount(shard.finding_count, 4096)) { + throw new APIRequestError("Fan-out execution shard is invalid", "INVALID_RESPONSE", 502); + } + return shard; + }); + return { ...value, shards } as FanoutExecutionView; +} diff --git a/web/src/api/schema.d.ts b/web/src/api/schema.d.ts index c9eb3c4f..03953ba3 100644 --- a/web/src/api/schema.d.ts +++ b/web/src/api/schema.d.ts @@ -836,6 +836,46 @@ export interface paths { patch?: never; trace?: never; }; + "/api/v1/runs/{run_id}/fanout-executions": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + /** + * List one Fan-out plan's execution history + * @description Returns the bounded shard execution history of one read-only Fan-out plan; shard rows carry only status, bounded counts, and model identity. + */ + get: operations["listRunFanoutExecutions"]; + put?: never; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; + "/api/v1/runs/{run_id}/fanout-executions/{execution_id}/cancel": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get?: never; + put?: never; + /** + * Cancel one read-only Fan-out execution + * @description Cancels one non-terminal read-only Fan-out execution under a fresh Run execution lease fence; terminal executions return their stored state unchanged. + */ + post: operations["cancelRunFanoutExecution"]; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/api/v1/runs/{run_id}/fanout-plans": { parameters: { query?: never; @@ -2824,6 +2864,10 @@ export interface components { token_upper_bound: number; tool_capability_grant: boolean; }; + FanoutExecutionCancelRequestView: { + confirm_cancel: boolean; + version: string; + }; FanoutExecutionShardView: { /** Format: int32 */ attempt_count: number; @@ -2868,6 +2912,11 @@ export interface components { /** Format: date-time */ updated_at: string; }; + FanoutExecutionsListView: { + items: components["schemas"]["FanoutExecutionView"][]; + plan_id: string; + protocol_version: string; + }; FanoutPlanView: { /** Format: date-time */ created_at: string; @@ -7255,6 +7304,89 @@ export interface operations { 500: components["responses"]["InternalError"]; }; }; + listRunFanoutExecutions: { + parameters: { + query?: { + /** @description Exact Fan-out plan identity */ + plan_id?: string; + }; + header?: never; + path: { + /** @description Run identity */ + run_id: string; + }; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description Successful read */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": { + data: components["schemas"]["FanoutExecutionsListView"]; + request_id: string; + /** @constant */ + version: "api.v1"; + }; + }; + }; + 400: components["responses"]["BadRequest"]; + 401: components["responses"]["Unauthorized"]; + 403: components["responses"]["Forbidden"]; + 404: components["responses"]["NotFound"]; + 414: components["responses"]["RequestTooLarge"]; + 429: components["responses"]["ResourceExhausted"]; + 500: components["responses"]["InternalError"]; + }; + }; + cancelRunFanoutExecution: { + parameters: { + query?: never; + header?: never; + path: { + /** @description Run identity */ + run_id: string; + /** @description Read-only Fan-out execution identity */ + execution_id: string; + }; + cookie?: never; + }; + requestBody: { + content: { + "application/json": components["schemas"]["FanoutExecutionCancelRequestView"]; + }; + }; + responses: { + /** @description Control request accepted or idempotently replayed */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": { + data: components["schemas"]["FanoutExecutionView"]; + request_id: string; + /** @constant */ + version: "api.v1"; + }; + }; + }; + 400: components["responses"]["BadRequest"]; + 401: components["responses"]["Unauthorized"]; + 403: components["responses"]["Forbidden"]; + 404: components["responses"]["NotFound"]; + 409: components["responses"]["Conflict"]; + 412: components["responses"]["FailedPrecondition"]; + 413: components["responses"]["RequestEntityTooLarge"]; + 414: components["responses"]["RequestTooLarge"]; + 415: components["responses"]["UnsupportedMediaType"]; + 429: components["responses"]["ResourceExhausted"]; + 500: components["responses"]["InternalError"]; + }; + }; listRunFanoutPlans: { parameters: { query?: { diff --git a/web/src/api/types.ts b/web/src/api/types.ts index 063c2772..3eee5bdc 100644 --- a/web/src/api/types.ts +++ b/web/src/api/types.ts @@ -21,6 +21,10 @@ export type EventView = components["schemas"]["EventView"]; export type ExternalSkillProjectionItemView = components["schemas"]["ExternalSkillProjectionItemView"]; export type ExternalSkillProjectionView = components["schemas"]["ExternalSkillProjectionView"]; export type FanoutPlanView = components["schemas"]["FanoutPlanView"]; +export type FanoutExecutionsListView = components["schemas"]["FanoutExecutionsListView"]; +export type FanoutExecutionView = components["schemas"]["FanoutExecutionView"]; +export type FanoutExecutionCancelRequestView = + components["schemas"]["FanoutExecutionCancelRequestView"]; export type FindingReportSummaryView = components["schemas"]["FindingReportSummaryView"]; export type FindingReportView = components["schemas"]["FindingReportView"]; export type HealthView = components["schemas"]["HealthView"]; diff --git a/web/src/components/run-projections.tsx b/web/src/components/run-projections.tsx index 3c174877..903a2f70 100644 --- a/web/src/components/run-projections.tsx +++ b/web/src/components/run-projections.tsx @@ -73,10 +73,36 @@ export function AgentGraphPanel({ client, runID }: ProjectionProps) { if (query.isLoading) return ; if (query.isError || !query.data) return ; if (query.data.nodes.length === 0) return 暂无 Agent; + const nodes = query.data.nodes; + // Render the child tree: the root first, its direct children indented + // beneath it (depth is bounded at 1 by the admission contract). + const ordered = [...nodes].sort((left, right) => { + if (left.id === query.data.root_agent_id) return -1; + if (right.id === query.data.root_agent_id) return 1; + return (left.parent_id ?? "").localeCompare(right.parent_id ?? "") || left.id.localeCompare(right.id); + }); + const root = ordered.find((node) => node.id === query.data.root_agent_id); return (
- {query.data.nodes.map((node) => ( -
+ {root && ( +
+
+ + {shortID(root.id)} + +
+
+ + + + + +
+
{root.skills.map((skill) => {skill})}
+
+ )} + {ordered.filter((node) => node.id !== query.data.root_agent_id).map((node) => ( +
{shortID(node.id)} @@ -87,6 +113,8 @@ export function AgentGraphPanel({ client, runID }: ProjectionProps) { + +
{node.skills.map((skill) => {skill})}
{node.completion && ( @@ -219,7 +247,7 @@ export function FanoutPanel({ client, runID }: ProjectionProps) { - {plan.latest_execution ? :
尚未执行
} +
))} void query.fetchNextPage()} /> @@ -227,6 +255,55 @@ export function FanoutPanel({ client, runID }: ProjectionProps) { ); } +function FanoutExecutions({ client, runID, planID }: { client: CyberAgentClient; runID: string; planID: string }) { + const { t } = useLocale(); + const queryClient = useQueryClient(); + const query = useQuery({ + queryKey: ["run", runID, "fanout-executions", planID], + queryFn: ({ signal }) => client.getRunFanoutExecutions(runID, planID, signal), + }); + const cancel = useMutation({ + mutationFn: (executionID: string) => client.cancelRunFanoutExecution(runID, executionID, { + version: "readonly_fanout_cancel.v1", confirm_cancel: true, + }, `web-fanout-cancel-${globalThis.crypto.randomUUID()}`), + onSuccess: () => { + void queryClient.invalidateQueries({ queryKey: ["run", runID, "fanout"] }); + void queryClient.invalidateQueries({ queryKey: ["run", runID, "fanout-executions", planID] }); + }, + }); + if (query.isLoading) return ; + if (query.isError || !query.data) return ; + if (query.data.items.length === 0) return
尚未执行
; + return ( +
+

{t("执行历史", "Execution history")}

+ {query.data.items.map((execution) => ( +
+
+ {shortID(execution.id)} + + {formatDate(execution.started_at)} + {client.hasControl && execution.status === "running" && ( + + )} +
+ + {execution.stop_code &&
{execution.stop_code}
} +
+ ))} + {cancel.error &&
+ {cancel.error instanceof Error ? cancel.error.message : t("取消 Fan-out 执行失败", "Fan-out execution cancellation failed")} +
} +
+ ); +} + function ShardTable({ execution }: { execution: NonNullable }) { const { t } = useLocale(); return (