Implement Hexis confirmations, disabled-by-default risk, and timeout=>unknown
Closes the biggest gap between the running execution engine and ECOSYSTEM-SPEC.md §4.3: confirmations were entirely unmodeled, so any capability could execute unconfirmed regardless of requires_confirmation. - New confirmations table + Confirmation domain type; POST /api/v1/confirmations mints a TTL-bound (120s) confirmation binding capability id+version, target entity, and a sorted-key args hash. - Execute() now requires a valid pending confirmation when the capability demands one: rejects missing, expired, consumed, or args/version-mismatched confirmations; consumes on success. - Capabilities gain enabled (destructive risk defaults to disabled, matching "must be turned on explicitly") and timeout_seconds. - One in-flight execution per (capability_id, target_entity_id); a second concurrent attempt is rejected (surfaced as 409 over HTTP). - Wall-clock timeout per capability now wraps the provider call; on timeout the outcome is "unknown" (new ExecutionStatus), never "failed", and the run is never auto-retried. - 9 new engine tests cover each guard from the spec's Phase 6 gate. Vikunja #274.
This commit is contained in:
kami
+123
-12
@@ -124,6 +124,23 @@ var migrations = []string{
|
||||
version INTEGER PRIMARY KEY,
|
||||
applied_at TEXT NOT NULL
|
||||
)`,
|
||||
`ALTER TABLE capabilities ADD COLUMN requires_confirmation INTEGER NOT NULL DEFAULT 0`,
|
||||
`ALTER TABLE capabilities ADD COLUMN enabled INTEGER NOT NULL DEFAULT 1`,
|
||||
`ALTER TABLE capabilities ADD COLUMN timeout_seconds INTEGER NOT NULL DEFAULT 30`,
|
||||
`ALTER TABLE executions ADD COLUMN confirmation_id TEXT NOT NULL DEFAULT ''`,
|
||||
`CREATE TABLE IF NOT EXISTS confirmations (
|
||||
id TEXT PRIMARY KEY,
|
||||
capability_id TEXT NOT NULL,
|
||||
capability_version INTEGER NOT NULL,
|
||||
target_entity_id TEXT NOT NULL,
|
||||
args_normalized TEXT NOT NULL DEFAULT '{}',
|
||||
args_hash TEXT NOT NULL,
|
||||
requester TEXT NOT NULL DEFAULT '',
|
||||
created_at TEXT NOT NULL,
|
||||
expires_at TEXT NOT NULL,
|
||||
state TEXT NOT NULL DEFAULT 'pending'
|
||||
)`,
|
||||
`CREATE INDEX IF NOT EXISTS idx_executions_inflight ON executions(capability_id, target_entity_id, status)`,
|
||||
}
|
||||
|
||||
const timeFmt = "2006-01-02T15:04:05.999999999Z07:00"
|
||||
@@ -153,9 +170,9 @@ func (s *Store) CreateCapability(c *domain.Capability) error {
|
||||
attrs, _ := json.Marshal(c.Attributes)
|
||||
|
||||
_, err := s.db.Exec(
|
||||
`INSERT INTO capabilities (id, name, description, target_types, target_entity_id, provider, operation, risk, read_only, expected_side_effects, attributes, created_at, updated_at, version)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
c.ID, c.Name, c.Description, string(targetTypes), c.TargetEntityID, c.Provider, c.Operation, c.Risk, boolInt(c.ReadOnly), c.ExpectedSideEffects, string(attrs), formatTime(c.CreatedAt), formatTime(c.UpdatedAt), c.Version,
|
||||
`INSERT INTO capabilities (id, name, description, target_types, target_entity_id, provider, operation, risk, read_only, expected_side_effects, requires_confirmation, enabled, timeout_seconds, attributes, created_at, updated_at, version)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
c.ID, c.Name, c.Description, string(targetTypes), c.TargetEntityID, c.Provider, c.Operation, c.Risk, boolInt(c.ReadOnly), c.ExpectedSideEffects, boolInt(c.RequiresConfirmation), boolInt(c.Enabled), c.TimeoutSeconds, string(attrs), formatTime(c.CreatedAt), formatTime(c.UpdatedAt), c.Version,
|
||||
)
|
||||
return err
|
||||
}
|
||||
@@ -165,12 +182,12 @@ func (s *Store) GetCapability(id string) (*domain.Capability, error) {
|
||||
defer s.mu.RUnlock()
|
||||
|
||||
row := s.db.QueryRow(
|
||||
`SELECT id, name, description, target_types, COALESCE(target_entity_id,''), provider, operation, risk, read_only, COALESCE(expected_side_effects,''), attributes, created_at, updated_at, version
|
||||
`SELECT id, name, description, target_types, COALESCE(target_entity_id,''), provider, operation, risk, read_only, COALESCE(expected_side_effects,''), requires_confirmation, enabled, timeout_seconds, attributes, created_at, updated_at, version
|
||||
FROM capabilities WHERE id = ?`, id,
|
||||
)
|
||||
c := &domain.Capability{}
|
||||
var targetTypes, attrs, createdAt, updatedAt string
|
||||
err := row.Scan(&c.ID, &c.Name, &c.Description, &targetTypes, &c.TargetEntityID, &c.Provider, &c.Operation, &c.Risk, &c.ReadOnly, &c.ExpectedSideEffects, &attrs, &createdAt, &updatedAt, &c.Version)
|
||||
err := row.Scan(&c.ID, &c.Name, &c.Description, &targetTypes, &c.TargetEntityID, &c.Provider, &c.Operation, &c.Risk, &c.ReadOnly, &c.ExpectedSideEffects, &c.RequiresConfirmation, &c.Enabled, &c.TimeoutSeconds, &attrs, &createdAt, &updatedAt, &c.Version)
|
||||
if err == sql.ErrNoRows {
|
||||
return nil, domain.ErrCapabilityNotFound
|
||||
}
|
||||
@@ -198,9 +215,9 @@ func (s *Store) UpdateCapability(c *domain.Capability) error {
|
||||
attrs, _ := json.Marshal(c.Attributes)
|
||||
|
||||
res, err := s.db.Exec(
|
||||
`UPDATE capabilities SET name=?, description=?, target_types=?, target_entity_id=?, provider=?, operation=?, risk=?, read_only=?, expected_side_effects=?, attributes=?, updated_at=?, version=version+1
|
||||
`UPDATE capabilities SET name=?, description=?, target_types=?, target_entity_id=?, provider=?, operation=?, risk=?, read_only=?, expected_side_effects=?, requires_confirmation=?, enabled=?, timeout_seconds=?, attributes=?, updated_at=?, version=version+1
|
||||
WHERE id=? AND version=?`,
|
||||
c.Name, c.Description, string(targetTypes), c.TargetEntityID, c.Provider, c.Operation, c.Risk, boolInt(c.ReadOnly), c.ExpectedSideEffects, string(attrs), formatTime(c.UpdatedAt), c.ID, c.Version,
|
||||
c.Name, c.Description, string(targetTypes), c.TargetEntityID, c.Provider, c.Operation, c.Risk, boolInt(c.ReadOnly), c.ExpectedSideEffects, boolInt(c.RequiresConfirmation), boolInt(c.Enabled), c.TimeoutSeconds, string(attrs), formatTime(c.UpdatedAt), c.ID, c.Version,
|
||||
)
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -217,7 +234,7 @@ func (s *Store) ListCapabilities(entityID string) ([]*domain.Capability, error)
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
|
||||
query := `SELECT id, name, description, target_types, COALESCE(target_entity_id,''), provider, operation, risk, read_only, COALESCE(expected_side_effects,''), attributes, created_at, updated_at, version
|
||||
query := `SELECT id, name, description, target_types, COALESCE(target_entity_id,''), provider, operation, risk, read_only, COALESCE(expected_side_effects,''), requires_confirmation, enabled, timeout_seconds, attributes, created_at, updated_at, version
|
||||
FROM capabilities`
|
||||
args := []any{}
|
||||
if entityID != "" {
|
||||
@@ -236,7 +253,7 @@ func (s *Store) ListCapabilities(entityID string) ([]*domain.Capability, error)
|
||||
for rows.Next() {
|
||||
c := &domain.Capability{}
|
||||
var targetTypes, attrs, createdAt, updatedAt string
|
||||
if err := rows.Scan(&c.ID, &c.Name, &c.Description, &targetTypes, &c.TargetEntityID, &c.Provider, &c.Operation, &c.Risk, &c.ReadOnly, &c.ExpectedSideEffects, &attrs, &createdAt, &updatedAt, &c.Version); err != nil {
|
||||
if err := rows.Scan(&c.ID, &c.Name, &c.Description, &targetTypes, &c.TargetEntityID, &c.Provider, &c.Operation, &c.Risk, &c.ReadOnly, &c.ExpectedSideEffects, &c.RequiresConfirmation, &c.Enabled, &c.TimeoutSeconds, &attrs, &createdAt, &updatedAt, &c.Version); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
json.Unmarshal([]byte(targetTypes), &c.TargetTypes)
|
||||
@@ -275,9 +292,9 @@ func (s *Store) CreateExecution(e *domain.Execution) error {
|
||||
evidence, _ := json.Marshal(e.ResolutionEvidence)
|
||||
|
||||
_, err := s.db.Exec(
|
||||
`INSERT INTO executions (id, capability_id, target_entity_id, entity_version, arguments, requested_by, origin, idempotency_key, status, result, error, resolution_evidence, correlation_id, created_at, updated_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
e.ID, e.CapabilityID, e.TargetEntityID, e.EntityVersion, string(args), string(reqBy), string(origin), nullString(e.IdempotencyKey), string(e.Status), string(result), e.Error, string(evidence), e.CorrelationID, formatTime(e.CreatedAt), formatTime(e.UpdatedAt),
|
||||
`INSERT INTO executions (id, capability_id, target_entity_id, entity_version, arguments, requested_by, origin, idempotency_key, confirmation_id, status, result, error, resolution_evidence, correlation_id, created_at, updated_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
e.ID, e.CapabilityID, e.TargetEntityID, e.EntityVersion, string(args), string(reqBy), string(origin), nullString(e.IdempotencyKey), e.ConfirmationID, string(e.Status), string(result), e.Error, string(evidence), e.CorrelationID, formatTime(e.CreatedAt), formatTime(e.UpdatedAt),
|
||||
)
|
||||
if err != nil {
|
||||
if strings.Contains(err.Error(), "UNIQUE") {
|
||||
@@ -375,6 +392,100 @@ func (s *Store) GetExecutionByIdempotencyKey(key string) (*domain.Execution, err
|
||||
return e, nil
|
||||
}
|
||||
|
||||
func (s *Store) GetInFlightExecution(capabilityID, targetEntityID string) (*domain.Execution, error) {
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
|
||||
row := s.db.QueryRow(
|
||||
`SELECT id FROM executions WHERE capability_id = ? AND target_entity_id = ? AND status = ? LIMIT 1`,
|
||||
capabilityID, targetEntityID, string(domain.ExecutionStarted),
|
||||
)
|
||||
var id string
|
||||
err := row.Scan(&id)
|
||||
if err == sql.ErrNoRows {
|
||||
return nil, domain.ErrExecutionNotFound
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return s.getExecutionLocked(id)
|
||||
}
|
||||
|
||||
// getExecutionLocked reads an execution without acquiring s.mu; callers must
|
||||
// already hold it (read or write).
|
||||
func (s *Store) getExecutionLocked(id string) (*domain.Execution, error) {
|
||||
row := s.db.QueryRow(
|
||||
`SELECT id, capability_id, target_entity_id, entity_version, arguments, requested_by, origin, COALESCE(idempotency_key,''), status, result, COALESCE(error,''), resolution_evidence, COALESCE(correlation_id,''), created_at, updated_at
|
||||
FROM executions WHERE id = ?`, id,
|
||||
)
|
||||
e := &domain.Execution{}
|
||||
var args, reqBy, origin, idempKey, status, result, errStr, evidence, corrID, createdAt, updatedAt string
|
||||
err := row.Scan(&e.ID, &e.CapabilityID, &e.TargetEntityID, &e.EntityVersion, &args, &reqBy, &origin, &idempKey, &status, &result, &errStr, &evidence, &corrID, &createdAt, &updatedAt)
|
||||
if err == sql.ErrNoRows {
|
||||
return nil, domain.ErrExecutionNotFound
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
json.Unmarshal([]byte(args), &e.Arguments)
|
||||
json.Unmarshal([]byte(reqBy), &e.RequestedBy)
|
||||
json.Unmarshal([]byte(origin), &e.Origin)
|
||||
json.Unmarshal([]byte(result), &e.Result)
|
||||
json.Unmarshal([]byte(evidence), &e.ResolutionEvidence)
|
||||
e.IdempotencyKey = idempKey
|
||||
e.Status = domain.ExecutionStatus(status)
|
||||
e.Error = errStr
|
||||
e.CorrelationID = corrID
|
||||
e.CreatedAt = parseTime(createdAt)
|
||||
e.UpdatedAt = parseTime(updatedAt)
|
||||
return e, nil
|
||||
}
|
||||
|
||||
// Confirmation operations
|
||||
|
||||
func (s *Store) CreateConfirmation(c *domain.Confirmation) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
_, err := s.db.Exec(
|
||||
`INSERT INTO confirmations (id, capability_id, capability_version, target_entity_id, args_normalized, args_hash, requester, created_at, expires_at, state)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
c.ID, c.CapabilityID, c.CapabilityVersion, c.TargetEntityID, c.ArgsNormalized, c.ArgsHash, c.Requester, formatTime(c.CreatedAt), formatTime(c.ExpiresAt), string(c.State),
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
func (s *Store) GetConfirmation(id string) (*domain.Confirmation, error) {
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
|
||||
row := s.db.QueryRow(
|
||||
`SELECT id, capability_id, capability_version, target_entity_id, args_normalized, args_hash, requester, created_at, expires_at, state
|
||||
FROM confirmations WHERE id = ?`, id,
|
||||
)
|
||||
c := &domain.Confirmation{}
|
||||
var createdAt, expiresAt, state string
|
||||
err := row.Scan(&c.ID, &c.CapabilityID, &c.CapabilityVersion, &c.TargetEntityID, &c.ArgsNormalized, &c.ArgsHash, &c.Requester, &createdAt, &expiresAt, &state)
|
||||
if err == sql.ErrNoRows {
|
||||
return nil, domain.ErrConfirmationNotFound
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
c.CreatedAt = parseTime(createdAt)
|
||||
c.ExpiresAt = parseTime(expiresAt)
|
||||
c.State = domain.ConfirmationState(state)
|
||||
return c, nil
|
||||
}
|
||||
|
||||
func (s *Store) UpdateConfirmationState(id string, state domain.ConfirmationState) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
_, err := s.db.Exec(`UPDATE confirmations SET state = ? WHERE id = ?`, string(state), id)
|
||||
return err
|
||||
}
|
||||
|
||||
// Event operations
|
||||
|
||||
func (s *Store) AppendEvent(evt *domain.Event) error {
|
||||
|
||||
Reference in New Issue
Block a user