Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
247 changes: 247 additions & 0 deletions docs/openapi.json
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down Expand Up @@ -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": {
Expand Down Expand Up @@ -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": {
Expand Down
2 changes: 2 additions & 0 deletions internal/app/api_command.go
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
75 changes: 75 additions & 0 deletions internal/application/readonly_fanout_execution.go
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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)
}
2 changes: 2 additions & 0 deletions internal/desktop/control_plane.go
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
Loading