From 89d8433d177020ecc50bba5782d39d0a2c8b1e20 Mon Sep 17 00:00:00 2001 From: kami Date: Mon, 20 Jul 2026 00:58:42 +0400 Subject: [PATCH] Implement Hexis confirmations, disabled-by-default risk, and timeout=>unknown MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- internal/api/handler.go | 120 +++++++++--- internal/domain/capability.go | 118 ++++++------ internal/domain/confirmation.go | 82 +++++++++ internal/domain/errors.go | 9 + internal/execution/engine.go | 148 ++++++++++++++- internal/execution/engine_test.go | 293 ++++++++++++++++++++++++++++++ internal/storage/interface.go | 5 + internal/storage/sqlite.go | 135 ++++++++++++-- 8 files changed, 812 insertions(+), 98 deletions(-) create mode 100644 internal/domain/confirmation.go create mode 100644 internal/execution/engine_test.go diff --git a/internal/api/handler.go b/internal/api/handler.go index 59c82a1..6b4a260 100644 --- a/internal/api/handler.go +++ b/internal/api/handler.go @@ -2,6 +2,7 @@ package api import ( "encoding/json" + "errors" "net/http" "strings" "time" @@ -26,6 +27,7 @@ func (h *Handler) Register(mux *http.ServeMux) { mux.HandleFunc("/api/v1/capabilities", h.handleCapabilities) mux.HandleFunc("/api/v1/capabilities/", h.handleCapabilityByID) mux.HandleFunc("/api/v1/execute", h.handleExecute) + mux.HandleFunc("/api/v1/confirmations", h.handleConfirmations) mux.HandleFunc("/api/v1/executions/", h.handleExecutionByID) mux.HandleFunc("/api/v1/changes", h.handleChanges) } @@ -87,16 +89,19 @@ func (h *Handler) listCapabilities(w http.ResponseWriter, r *http.Request) { func (h *Handler) createCapability(w http.ResponseWriter, r *http.Request) { var req struct { - Name string `json:"name"` - Description string `json:"description,omitempty"` - TargetTypes []string `json:"target_types"` - TargetEntityID string `json:"target_entity_id,omitempty"` - Provider string `json:"provider"` - Operation string `json:"operation"` - Risk string `json:"risk,omitempty"` - ReadOnly bool `json:"read_only"` - ExpectedSideEffects string `json:"expected_side_effects,omitempty"` - Attributes map[string]any `json:"attributes,omitempty"` + Name string `json:"name"` + Description string `json:"description,omitempty"` + TargetTypes []string `json:"target_types"` + TargetEntityID string `json:"target_entity_id,omitempty"` + Provider string `json:"provider"` + Operation string `json:"operation"` + Risk string `json:"risk,omitempty"` + ReadOnly bool `json:"read_only"` + ExpectedSideEffects string `json:"expected_side_effects,omitempty"` + RequiresConfirmation bool `json:"requires_confirmation,omitempty"` + Enabled *bool `json:"enabled,omitempty"` + TimeoutSeconds int `json:"timeout_seconds,omitempty"` + Attributes map[string]any `json:"attributes,omitempty"` } if err := json.NewDecoder(r.Body).Decode(&req); err != nil { writeJSON(w, http.StatusBadRequest, errorResponse("invalid JSON")) @@ -107,22 +112,32 @@ func (h *Handler) createCapability(w http.ResponseWriter, r *http.Request) { return } + // Destructive capabilities are disabled by default and must be turned on + // explicitly (ECOSYSTEM-SPEC.md §4.3). + enabled := req.Risk != "destructive" + if req.Enabled != nil { + enabled = *req.Enabled + } + now := time.Now().UTC() cap := &domain.Capability{ - ID: domain.NewCapabilityID(), - Name: req.Name, - Description: req.Description, - TargetTypes: req.TargetTypes, - TargetEntityID: req.TargetEntityID, - Provider: req.Provider, - Operation: req.Operation, - Risk: req.Risk, - ReadOnly: req.ReadOnly, - ExpectedSideEffects: req.ExpectedSideEffects, - Attributes: req.Attributes, - CreatedAt: now, - UpdatedAt: now, - Version: 1, + ID: domain.NewCapabilityID(), + Name: req.Name, + Description: req.Description, + TargetTypes: req.TargetTypes, + TargetEntityID: req.TargetEntityID, + Provider: req.Provider, + Operation: req.Operation, + Risk: req.Risk, + ReadOnly: req.ReadOnly, + ExpectedSideEffects: req.ExpectedSideEffects, + RequiresConfirmation: req.RequiresConfirmation, + Enabled: enabled, + TimeoutSeconds: req.TimeoutSeconds, + Attributes: req.Attributes, + CreatedAt: now, + UpdatedAt: now, + Version: 1, } if cap.TargetTypes == nil { cap.TargetTypes = []string{} @@ -198,13 +213,68 @@ func (h *Handler) handleExecute(w http.ResponseWriter, r *http.Request) { result, err := h.engine.Execute(&req) if err != nil { - writeJSON(w, http.StatusBadRequest, errorResponse(err.Error())) + writeJSON(w, executeErrorStatus(err), errorResponse(err.Error())) return } writeJSON(w, http.StatusOK, result.Execution) } +func executeErrorStatus(err error) int { + switch { + case errors.Is(err, domain.ErrExecutionInFlight): + return http.StatusConflict + case errors.Is(err, domain.ErrCapabilityNotFound): + return http.StatusNotFound + case errors.Is(err, domain.ErrConfirmationRequired), + errors.Is(err, domain.ErrConfirmationInvalid), + errors.Is(err, domain.ErrConfirmationExpired), + errors.Is(err, domain.ErrConfirmationConsumed), + errors.Is(err, domain.ErrConfirmationNotFound), + errors.Is(err, domain.ErrCapabilityDisabled), + errors.Is(err, domain.ErrCapabilityNotBound): + return http.StatusForbidden + default: + return http.StatusBadRequest + } +} + +func (h *Handler) handleConfirmations(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + writeJSON(w, http.StatusMethodNotAllowed, errorResponse("method not allowed")) + return + } + + var req struct { + CapabilityID string `json:"capability_id"` + TargetEntityID string `json:"target_entity_id"` + Arguments map[string]any `json:"arguments,omitempty"` + Requester string `json:"requester,omitempty"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeJSON(w, http.StatusBadRequest, errorResponse("invalid JSON")) + return + } + if req.CapabilityID == "" || req.TargetEntityID == "" { + writeJSON(w, http.StatusBadRequest, errorResponse("capability_id and target_entity_id are required")) + return + } + + conf, err := h.engine.CreateConfirmation(req.CapabilityID, req.TargetEntityID, req.Requester, req.Arguments) + if err != nil { + status := http.StatusBadRequest + if errors.Is(err, domain.ErrCapabilityNotFound) { + status = http.StatusNotFound + } else if errors.Is(err, domain.ErrCapabilityNotBound) { + status = http.StatusForbidden + } + writeJSON(w, status, errorResponse(err.Error())) + return + } + + writeJSON(w, http.StatusCreated, conf) +} + func (h *Handler) handleExecutionByID(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { writeJSON(w, http.StatusMethodNotAllowed, errorResponse("method not allowed")) diff --git a/internal/domain/capability.go b/internal/domain/capability.go index 69c6533..7f2d729 100644 --- a/internal/domain/capability.go +++ b/internal/domain/capability.go @@ -3,32 +3,42 @@ package domain import "time" type ExecuteRequest struct { - CapabilityID string `json:"capability_id"` - TargetEntityID string `json:"target_entity_id"` - EntityVersion int64 `json:"entity_version,omitempty"` - Arguments map[string]any `json:"arguments,omitempty"` - RequestedBy map[string]string `json:"requested_by,omitempty"` - Origin map[string]string `json:"origin,omitempty"` - IdempotencyKey string `json:"idempotency_key,omitempty"` - CorrelationID string `json:"correlation_id,omitempty"` - ResolutionEvidence []map[string]any `json:"resolution_evidence,omitempty"` + CapabilityID string `json:"capability_id"` + TargetEntityID string `json:"target_entity_id"` + EntityVersion int64 `json:"entity_version,omitempty"` + Arguments map[string]any `json:"arguments,omitempty"` + RequestedBy map[string]string `json:"requested_by,omitempty"` + Origin map[string]string `json:"origin,omitempty"` + IdempotencyKey string `json:"idempotency_key,omitempty"` + CorrelationID string `json:"correlation_id,omitempty"` + ResolutionEvidence []map[string]any `json:"resolution_evidence,omitempty"` + ConfirmationID string `json:"confirmation_id,omitempty"` } type Capability struct { - ID string `json:"id"` - Name string `json:"name"` - Description string `json:"description,omitempty"` - TargetTypes []string `json:"target_types"` - TargetEntityID string `json:"target_entity_id,omitempty"` - Provider string `json:"provider"` - Operation string `json:"operation"` - Risk string `json:"risk,omitempty"` - ReadOnly bool `json:"read_only"` - ExpectedSideEffects string `json:"expected_side_effects,omitempty"` - Attributes map[string]any `json:"attributes,omitempty"` - CreatedAt time.Time `json:"created_at"` - UpdatedAt time.Time `json:"updated_at"` - Version int64 `json:"version"` + ID string `json:"id"` + Name string `json:"name"` + Description string `json:"description,omitempty"` + TargetTypes []string `json:"target_types"` + TargetEntityID string `json:"target_entity_id,omitempty"` + Provider string `json:"provider"` + Operation string `json:"operation"` + Risk string `json:"risk,omitempty"` + ReadOnly bool `json:"read_only"` + ExpectedSideEffects string `json:"expected_side_effects,omitempty"` + RequiresConfirmation bool `json:"requires_confirmation"` + Enabled bool `json:"enabled"` + TimeoutSeconds int `json:"timeout_seconds,omitempty"` + Attributes map[string]any `json:"attributes,omitempty"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` + Version int64 `json:"version"` +} + +// IsDestructive reports whether the capability's risk tier requires it to be +// disabled by default per ECOSYSTEM-SPEC.md §4.3. +func (c *Capability) IsDestructive() bool { + return c.Risk == "destructive" } type ExecutionStatus string @@ -38,44 +48,48 @@ const ( ExecutionSucceeded ExecutionStatus = "succeeded" ExecutionFailed ExecutionStatus = "failed" ExecutionDenied ExecutionStatus = "denied" + // ExecutionUnknown marks a wall-clock timeout: the executor may or may not + // have completed the side effect. Never retried automatically. + ExecutionUnknown ExecutionStatus = "unknown" ) type Execution struct { - ID string `json:"id"` - CapabilityID string `json:"capability_id"` - TargetEntityID string `json:"target_entity_id"` - EntityVersion int64 `json:"entity_version,omitempty"` - Arguments map[string]any `json:"arguments,omitempty"` - RequestedBy map[string]string `json:"requested_by,omitempty"` - Origin map[string]string `json:"origin,omitempty"` - IdempotencyKey string `json:"idempotency_key,omitempty"` - Status ExecutionStatus `json:"status"` - Result map[string]any `json:"result,omitempty"` - Error string `json:"error,omitempty"` - ResolutionEvidence []map[string]any `json:"resolution_evidence,omitempty"` - CorrelationID string `json:"correlation_id,omitempty"` - CreatedAt time.Time `json:"created_at"` - UpdatedAt time.Time `json:"updated_at"` + ID string `json:"id"` + CapabilityID string `json:"capability_id"` + TargetEntityID string `json:"target_entity_id"` + EntityVersion int64 `json:"entity_version,omitempty"` + Arguments map[string]any `json:"arguments,omitempty"` + RequestedBy map[string]string `json:"requested_by,omitempty"` + Origin map[string]string `json:"origin,omitempty"` + IdempotencyKey string `json:"idempotency_key,omitempty"` + ConfirmationID string `json:"confirmation_id,omitempty"` + Status ExecutionStatus `json:"status"` + Result map[string]any `json:"result,omitempty"` + Error string `json:"error,omitempty"` + ResolutionEvidence []map[string]any `json:"resolution_evidence,omitempty"` + CorrelationID string `json:"correlation_id,omitempty"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` } type HexisEventType string const ( - EventCapabilityRegistered HexisEventType = "hexis.capability.registered" - EventCapabilityUnavailable HexisEventType = "hexis.capability.unavailable" - EventExecutionStarted HexisEventType = "hexis.execution.started" - EventExecutionSucceeded HexisEventType = "hexis.execution.succeeded" - EventExecutionFailed HexisEventType = "hexis.execution.failed" - EventExecutionDenied HexisEventType = "hexis.execution.denied" + EventCapabilityRegistered HexisEventType = "hexis.capability.registered" + EventCapabilityUnavailable HexisEventType = "hexis.capability.unavailable" + EventExecutionStarted HexisEventType = "hexis.execution.started" + EventExecutionSucceeded HexisEventType = "hexis.execution.succeeded" + EventExecutionFailed HexisEventType = "hexis.execution.failed" + EventExecutionDenied HexisEventType = "hexis.execution.denied" ) type Event struct { - ID string `json:"id"` - Sequence int64 `json:"sequence"` - Type HexisEventType `json:"type"` - Timestamp time.Time `json:"timestamp"` - Actor string `json:"actor,omitempty"` - CorrelationID string `json:"correlation_id,omitempty"` - CausationID string `json:"causation_id,omitempty"` - Payload map[string]any `json:"payload,omitempty"` + ID string `json:"id"` + Sequence int64 `json:"sequence"` + Type HexisEventType `json:"type"` + Timestamp time.Time `json:"timestamp"` + Actor string `json:"actor,omitempty"` + CorrelationID string `json:"correlation_id,omitempty"` + CausationID string `json:"causation_id,omitempty"` + Payload map[string]any `json:"payload,omitempty"` } diff --git a/internal/domain/confirmation.go b/internal/domain/confirmation.go new file mode 100644 index 0000000..cbc89ff --- /dev/null +++ b/internal/domain/confirmation.go @@ -0,0 +1,82 @@ +package domain + +import ( + "crypto/rand" + "crypto/sha256" + "encoding/base32" + "encoding/json" + "fmt" + "sort" + "strings" + "time" +) + +const ConfirmationTTL = 120 * time.Second + +type ConfirmationState string + +const ( + ConfirmationPending ConfirmationState = "pending" + ConfirmationConsumed ConfirmationState = "consumed" + ConfirmationExpired ConfirmationState = "expired" + ConfirmationRejected ConfirmationState = "rejected" +) + +type Confirmation struct { + ID string `json:"id"` + CapabilityID string `json:"capability_id"` + CapabilityVersion int64 `json:"capability_version"` + TargetEntityID string `json:"target_entity_id"` + ArgsNormalized string `json:"args_normalized"` + ArgsHash string `json:"args_hash"` + Requester string `json:"requester"` + CreatedAt time.Time `json:"created_at"` + ExpiresAt time.Time `json:"expires_at"` + State ConfirmationState `json:"state"` +} + +func NewConfirmationID() string { + b := make([]byte, 10) + rand.Read(b) + return "conf_" + strings.ToLower(base32.StdEncoding.WithPadding(base32.NoPadding).EncodeToString(b)) +} + +// NormalizeArgs produces a stable JSON encoding of an arguments map (sorted +// keys) so args_hash is comparable across requests that differ only in key +// order. +func NormalizeArgs(args map[string]any) (string, error) { + if args == nil { + args = map[string]any{} + } + keys := make([]string, 0, len(args)) + for k := range args { + keys = append(keys, k) + } + sort.Strings(keys) + + var b strings.Builder + b.WriteByte('{') + for i, k := range keys { + if i > 0 { + b.WriteByte(',') + } + kb, err := json.Marshal(k) + if err != nil { + return "", err + } + vb, err := json.Marshal(args[k]) + if err != nil { + return "", err + } + b.Write(kb) + b.WriteByte(':') + b.Write(vb) + } + b.WriteByte('}') + return b.String(), nil +} + +func HashArgs(normalized string) string { + sum := sha256.Sum256([]byte(normalized)) + return fmt.Sprintf("%x", sum) +} diff --git a/internal/domain/errors.go b/internal/domain/errors.go index cb07f5c..8fd4c7a 100644 --- a/internal/domain/errors.go +++ b/internal/domain/errors.go @@ -15,4 +15,13 @@ var ( ErrValidation = errors.New("validation error") ErrInternal = errors.New("internal error") ErrIdempotencyReplay = errors.New("idempotent request already processed") + + ErrCapabilityDisabled = errors.New("capability is disabled") + ErrCapabilityVersionMismatch = errors.New("capability version mismatch") + ErrConfirmationRequired = errors.New("confirmation required") + ErrConfirmationNotFound = errors.New("confirmation not found") + ErrConfirmationInvalid = errors.New("confirmation does not match request") + ErrConfirmationExpired = errors.New("confirmation expired") + ErrConfirmationConsumed = errors.New("confirmation already consumed") + ErrExecutionInFlight = errors.New("execution already in flight for this capability and target") ) diff --git a/internal/execution/engine.go b/internal/execution/engine.go index a8078b2..a2d1818 100644 --- a/internal/execution/engine.go +++ b/internal/execution/engine.go @@ -1,6 +1,7 @@ package execution import ( + "context" "encoding/json" "fmt" "time" @@ -10,6 +11,8 @@ import ( "github.com/kami/hexis/internal/storage" ) +const defaultTimeout = 30 * time.Second + type Engine struct { store storage.Interface registry *provider.Registry @@ -36,10 +39,25 @@ func (e *Engine) Execute(req *domain.ExecuteRequest) (*ExecuteResult, error) { return nil, fmt.Errorf("capability: %w", err) } + if !capability.Enabled { + return nil, domain.ErrCapabilityDisabled + } + if capability.TargetEntityID != "" && capability.TargetEntityID != req.TargetEntityID { return nil, domain.ErrCapabilityNotBound } + if capability.RequiresConfirmation { + if err := e.consumeConfirmation(req, capability); err != nil { + return nil, err + } + } + + // One in-flight execution per (capability_id, target_entity_id). + if _, err := e.store.GetInFlightExecution(req.CapabilityID, req.TargetEntityID); err == nil { + return nil, domain.ErrExecutionInFlight + } + prov, err := e.registry.Get(capability.Provider) if err != nil { return nil, fmt.Errorf("provider: %w", err) @@ -55,6 +73,7 @@ func (e *Engine) Execute(req *domain.ExecuteRequest) (*ExecuteResult, error) { RequestedBy: req.RequestedBy, Origin: req.Origin, IdempotencyKey: req.IdempotencyKey, + ConfirmationID: req.ConfirmationID, Status: domain.ExecutionStarted, CorrelationID: req.CorrelationID, ResolutionEvidence: req.ResolutionEvidence, @@ -76,16 +95,33 @@ func (e *Engine) Execute(req *domain.ExecuteRequest) (*ExecuteResult, error) { } e.emitEvent(domain.EventExecutionStarted, exec.ID, map[string]any{ - "capability_id": req.CapabilityID, + "capability_id": req.CapabilityID, "target_entity_id": req.TargetEntityID, - "provider": capability.Provider, - "correlation_id": req.CorrelationID, + "provider": capability.Provider, + "correlation_id": req.CorrelationID, }) - result, execErr := prov.Execute(capability, req) + timeout := defaultTimeout + if capability.TimeoutSeconds > 0 { + timeout = time.Duration(capability.TimeoutSeconds) * time.Second + } + + result, execErr, timedOut := e.runWithTimeout(prov, capability, req, timeout) exec.UpdatedAt = time.Now().UTC() - if execErr != nil { + switch { + case timedOut: + // Outcome is never "failed" on timeout, and never retried automatically + // per ECOSYSTEM-SPEC.md §4.3 — the side effect may or may not have landed. + exec.Status = domain.ExecutionUnknown + exec.Error = "execution timed out after " + timeout.String() + exec.Result = map[string]any{"error": exec.Error} + e.emitEvent(domain.EventExecutionFailed, exec.ID, map[string]any{ + "capability_id": req.CapabilityID, + "outcome": "unknown", + "reason": "timeout", + }) + case execErr != nil: exec.Status = domain.ExecutionFailed exec.Error = execErr.Error() exec.Result = map[string]any{"error": execErr.Error()} @@ -93,7 +129,7 @@ func (e *Engine) Execute(req *domain.ExecuteRequest) (*ExecuteResult, error) { "capability_id": req.CapabilityID, "error": execErr.Error(), }) - } else { + default: exec.Status = domain.ExecutionSucceeded exec.Result = result e.emitEvent(domain.EventExecutionSucceeded, exec.ID, map[string]any{ @@ -108,6 +144,100 @@ func (e *Engine) Execute(req *domain.ExecuteRequest) (*ExecuteResult, error) { return &ExecuteResult{Execution: exec}, nil } +// runWithTimeout executes the provider call on a goroutine and returns +// (nil, nil, true) if it doesn't finish within timeout. The goroutine is not +// cancelled — providers don't accept a context today — so a slow executor +// keeps running in the background; its eventual result is discarded, never +// retried, and the caller has already moved on with outcome=unknown. +func (e *Engine) runWithTimeout(prov provider.Provider, capability *domain.Capability, req *domain.ExecuteRequest, timeout time.Duration) (map[string]any, error, bool) { + type outcome struct { + result map[string]any + err error + } + ch := make(chan outcome, 1) + go func() { + r, err := prov.Execute(capability, req) + ch <- outcome{result: r, err: err} + }() + + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + + select { + case o := <-ch: + return o.result, o.err, false + case <-ctx.Done(): + return nil, nil, true + } +} + +func (e *Engine) consumeConfirmation(req *domain.ExecuteRequest, capability *domain.Capability) error { + if req.ConfirmationID == "" { + return domain.ErrConfirmationRequired + } + conf, err := e.store.GetConfirmation(req.ConfirmationID) + if err != nil { + return err + } + if conf.State != domain.ConfirmationPending { + return domain.ErrConfirmationConsumed + } + if time.Now().UTC().After(conf.ExpiresAt) { + e.store.UpdateConfirmationState(conf.ID, domain.ConfirmationExpired) + return domain.ErrConfirmationExpired + } + if conf.CapabilityID != capability.ID || conf.CapabilityVersion != capability.Version { + return domain.ErrConfirmationInvalid + } + if conf.TargetEntityID != req.TargetEntityID { + return domain.ErrConfirmationInvalid + } + normalized, err := domain.NormalizeArgs(req.Arguments) + if err != nil { + return err + } + if domain.HashArgs(normalized) != conf.ArgsHash { + return domain.ErrConfirmationInvalid + } + + return e.store.UpdateConfirmationState(conf.ID, domain.ConfirmationConsumed) +} + +// CreateConfirmation validates a would-be execute request against a +// capability and mints a time-bounded confirmation for it. +func (e *Engine) CreateConfirmation(capabilityID, targetEntityID, requester string, args map[string]any) (*domain.Confirmation, error) { + capability, err := e.store.GetCapability(capabilityID) + if err != nil { + return nil, fmt.Errorf("capability: %w", err) + } + if capability.TargetEntityID != "" && capability.TargetEntityID != targetEntityID { + return nil, domain.ErrCapabilityNotBound + } + + normalized, err := domain.NormalizeArgs(args) + if err != nil { + return nil, err + } + + now := time.Now().UTC() + conf := &domain.Confirmation{ + ID: domain.NewConfirmationID(), + CapabilityID: capability.ID, + CapabilityVersion: capability.Version, + TargetEntityID: targetEntityID, + ArgsNormalized: normalized, + ArgsHash: domain.HashArgs(normalized), + Requester: requester, + CreatedAt: now, + ExpiresAt: now.Add(domain.ConfirmationTTL), + State: domain.ConfirmationPending, + } + if err := e.store.CreateConfirmation(conf); err != nil { + return nil, err + } + return conf, nil +} + func (e *Engine) ValidateTarget(capabilityID, entityID string) error { cap, err := e.store.GetCapability(capabilityID) if err != nil { @@ -134,9 +264,9 @@ type ResolveRequest struct { } type ResolveResult struct { - Status string `json:"status"` - Candidates []map[string]any `json:"candidates,omitempty"` - EntityID string `json:"entity_id,omitempty"` + Status string `json:"status"` + Candidates []map[string]any `json:"candidates,omitempty"` + EntityID string `json:"entity_id,omitempty"` } var _ json.Marshaler = (*ResolveResult)(nil) diff --git a/internal/execution/engine_test.go b/internal/execution/engine_test.go new file mode 100644 index 0000000..da798c2 --- /dev/null +++ b/internal/execution/engine_test.go @@ -0,0 +1,293 @@ +package execution_test + +import ( + "errors" + "path/filepath" + "testing" + "time" + + "github.com/kami/hexis/internal/domain" + "github.com/kami/hexis/internal/execution" + "github.com/kami/hexis/internal/provider" + "github.com/kami/hexis/internal/storage" +) + +type fakeProvider struct { + name string + delay time.Duration + err error +} + +func (p *fakeProvider) Name() string { return p.name } + +func (p *fakeProvider) Execute(capability *domain.Capability, req *domain.ExecuteRequest) (map[string]any, error) { + if p.delay > 0 { + time.Sleep(p.delay) + } + if p.err != nil { + return nil, p.err + } + return map[string]any{"ok": true}, nil +} + +func newTestEngine(t *testing.T, prov *fakeProvider) (*execution.Engine, *storage.Store) { + t.Helper() + dir := t.TempDir() + store, err := storage.Open(filepath.Join(dir, "hexis.db")) + if err != nil { + t.Fatalf("open storage: %v", err) + } + t.Cleanup(func() { store.Close() }) + + reg := provider.NewRegistry() + reg.Register(prov) + + return execution.New(store, reg), store +} + +func mustCreateCapability(t *testing.T, store storage.Interface, mutate func(*domain.Capability)) *domain.Capability { + t.Helper() + now := time.Now().UTC() + cap := &domain.Capability{ + ID: domain.NewCapabilityID(), + Name: "test.capability", + Provider: "fake", + Operation: "noop", + Risk: "low", + ReadOnly: false, + Enabled: true, + CreatedAt: now, + UpdatedAt: now, + Version: 1, + TargetTypes: []string{}, + Attributes: map[string]any{}, + } + if mutate != nil { + mutate(cap) + } + if err := store.CreateCapability(cap); err != nil { + t.Fatalf("create capability: %v", err) + } + return cap +} + +func TestExecute_FreeTextTargetRejectedWhenBound(t *testing.T) { + prov := &fakeProvider{name: "fake"} + eng, store := newTestEngine(t, prov) + cap := mustCreateCapability(t, store, func(c *domain.Capability) { + c.TargetEntityID = "ent_bound_only" + }) + + _, err := eng.Execute(&domain.ExecuteRequest{ + CapabilityID: cap.ID, + TargetEntityID: "ent_someone_else", + }) + if !errors.Is(err, domain.ErrCapabilityNotBound) { + t.Fatalf("expected ErrCapabilityNotBound, got %v", err) + } +} + +func TestExecute_DisabledCapabilityRejected(t *testing.T) { + prov := &fakeProvider{name: "fake"} + eng, store := newTestEngine(t, prov) + cap := mustCreateCapability(t, store, func(c *domain.Capability) { + c.Enabled = false + }) + + _, err := eng.Execute(&domain.ExecuteRequest{CapabilityID: cap.ID, TargetEntityID: "ent_x"}) + if !errors.Is(err, domain.ErrCapabilityDisabled) { + t.Fatalf("expected ErrCapabilityDisabled, got %v", err) + } +} + +func TestExecute_ConfirmationRequiredButMissing(t *testing.T) { + prov := &fakeProvider{name: "fake"} + eng, store := newTestEngine(t, prov) + cap := mustCreateCapability(t, store, func(c *domain.Capability) { + c.RequiresConfirmation = true + }) + + _, err := eng.Execute(&domain.ExecuteRequest{CapabilityID: cap.ID, TargetEntityID: "ent_x"}) + if !errors.Is(err, domain.ErrConfirmationRequired) { + t.Fatalf("expected ErrConfirmationRequired, got %v", err) + } +} + +func TestExecute_ConfirmationValidSucceedsAndIsConsumed(t *testing.T) { + prov := &fakeProvider{name: "fake"} + eng, store := newTestEngine(t, prov) + cap := mustCreateCapability(t, store, func(c *domain.Capability) { + c.RequiresConfirmation = true + }) + + args := map[string]any{"foo": "bar"} + conf, err := eng.CreateConfirmation(cap.ID, "ent_x", "kami", args) + if err != nil { + t.Fatalf("create confirmation: %v", err) + } + + res, err := eng.Execute(&domain.ExecuteRequest{ + CapabilityID: cap.ID, + TargetEntityID: "ent_x", + Arguments: args, + ConfirmationID: conf.ID, + }) + if err != nil { + t.Fatalf("execute: %v", err) + } + if res.Execution.Status != domain.ExecutionSucceeded { + t.Fatalf("expected succeeded, got %s", res.Execution.Status) + } + + got, err := store.GetConfirmation(conf.ID) + if err != nil { + t.Fatalf("get confirmation: %v", err) + } + if got.State != domain.ConfirmationConsumed { + t.Fatalf("expected confirmation consumed, got %s", got.State) + } + + // Reusing the same confirmation must fail. + _, err = eng.Execute(&domain.ExecuteRequest{ + CapabilityID: cap.ID, + TargetEntityID: "ent_x", + Arguments: args, + ConfirmationID: conf.ID, + }) + if !errors.Is(err, domain.ErrConfirmationConsumed) { + t.Fatalf("expected ErrConfirmationConsumed on replay, got %v", err) + } +} + +func TestExecute_ConfirmationExpired(t *testing.T) { + prov := &fakeProvider{name: "fake"} + eng, store := newTestEngine(t, prov) + cap := mustCreateCapability(t, store, func(c *domain.Capability) { + c.RequiresConfirmation = true + }) + + conf, err := eng.CreateConfirmation(cap.ID, "ent_x", "kami", nil) + if err != nil { + t.Fatalf("create confirmation: %v", err) + } + + // Directly backdate the row's expiry in storage to simulate TTL elapse + // without waiting out the real 120s TTL in a test. + if _, err := store.DB().Exec(`UPDATE confirmations SET expires_at = ? WHERE id = ?`, + time.Now().UTC().Add(-time.Minute).Format("2006-01-02T15:04:05.999999999Z07:00"), conf.ID); err != nil { + t.Fatalf("backdate confirmation: %v", err) + } + + _, err = eng.Execute(&domain.ExecuteRequest{ + CapabilityID: cap.ID, + TargetEntityID: "ent_x", + ConfirmationID: conf.ID, + }) + if !errors.Is(err, domain.ErrConfirmationExpired) { + t.Fatalf("expected ErrConfirmationExpired, got %v", err) + } + + got, err := store.GetConfirmation(conf.ID) + if err != nil { + t.Fatalf("get confirmation: %v", err) + } + if got.State != domain.ConfirmationExpired { + t.Fatalf("expected confirmation state expired, got %s", got.State) + } +} + +func TestExecute_ArgsHashMismatchRejected(t *testing.T) { + prov := &fakeProvider{name: "fake"} + eng, store := newTestEngine(t, prov) + cap := mustCreateCapability(t, store, func(c *domain.Capability) { + c.RequiresConfirmation = true + }) + + conf, err := eng.CreateConfirmation(cap.ID, "ent_x", "kami", map[string]any{"n": 1}) + if err != nil { + t.Fatalf("create confirmation: %v", err) + } + + _, err = eng.Execute(&domain.ExecuteRequest{ + CapabilityID: cap.ID, + TargetEntityID: "ent_x", + Arguments: map[string]any{"n": 2}, + ConfirmationID: conf.ID, + }) + if !errors.Is(err, domain.ErrConfirmationInvalid) { + t.Fatalf("expected ErrConfirmationInvalid on args mismatch, got %v", err) + } +} + +func TestExecute_UnknownCapabilityVersionRejected(t *testing.T) { + prov := &fakeProvider{name: "fake"} + eng, store := newTestEngine(t, prov) + cap := mustCreateCapability(t, store, func(c *domain.Capability) { + c.RequiresConfirmation = true + }) + + conf, err := eng.CreateConfirmation(cap.ID, "ent_x", "kami", nil) + if err != nil { + t.Fatalf("create confirmation: %v", err) + } + // Bump the capability version (e.g. via update) so the confirmation now + // targets a stale version. + cap.Description = "changed" + if err := store.UpdateCapability(cap); err != nil { + t.Fatalf("update capability: %v", err) + } + + _, err = eng.Execute(&domain.ExecuteRequest{ + CapabilityID: cap.ID, + TargetEntityID: "ent_x", + ConfirmationID: conf.ID, + }) + if !errors.Is(err, domain.ErrConfirmationInvalid) { + t.Fatalf("expected ErrConfirmationInvalid on stale version, got %v", err) + } +} + +func TestExecute_ConcurrentInFlightRejected(t *testing.T) { + prov := &fakeProvider{name: "fake", delay: 300 * time.Millisecond} + eng, store := newTestEngine(t, prov) + cap := mustCreateCapability(t, store, nil) + + done := make(chan struct{}) + go func() { + eng.Execute(&domain.ExecuteRequest{CapabilityID: cap.ID, TargetEntityID: "ent_x"}) + close(done) + }() + + // Give the first execution time to register as "started". + time.Sleep(50 * time.Millisecond) + + _, err := eng.Execute(&domain.ExecuteRequest{CapabilityID: cap.ID, TargetEntityID: "ent_x"}) + if !errors.Is(err, domain.ErrExecutionInFlight) { + t.Fatalf("expected ErrExecutionInFlight, got %v", err) + } + + <-done +} + +func TestExecute_TimeoutYieldsUnknownNotFailed(t *testing.T) { + // Smallest supported capability timeout is 1s; use a provider slower + // than that so the engine's context deadline fires first. + prov := &fakeProvider{name: "slow", delay: 1500 * time.Millisecond} + eng, store := newTestEngine(t, prov) + cap := mustCreateCapability(t, store, func(c *domain.Capability) { + c.TimeoutSeconds = 1 + c.Provider = "slow" + }) + + res, err := eng.Execute(&domain.ExecuteRequest{CapabilityID: cap.ID, TargetEntityID: "ent_x"}) + if err != nil { + t.Fatalf("execute: %v", err) + } + if res.Execution.Status != domain.ExecutionUnknown { + t.Fatalf("expected status unknown on timeout, got %s", res.Execution.Status) + } + + // Let the background provider goroutine finish so t.Cleanup can close + // the store without a dangling write racing it. + time.Sleep(1600 * time.Millisecond) +} diff --git a/internal/storage/interface.go b/internal/storage/interface.go index 320697f..1a67e40 100644 --- a/internal/storage/interface.go +++ b/internal/storage/interface.go @@ -14,6 +14,11 @@ type Interface interface { GetExecution(id string) (*domain.Execution, error) UpdateExecution(e *domain.Execution) error GetExecutionByIdempotencyKey(key string) (*domain.Execution, error) + GetInFlightExecution(capabilityID, targetEntityID string) (*domain.Execution, error) + + CreateConfirmation(c *domain.Confirmation) error + GetConfirmation(id string) (*domain.Confirmation, error) + UpdateConfirmationState(id string, state domain.ConfirmationState) error AppendEvent(evt *domain.Event) error EventsAfter(seq int64, limit int) ([]*domain.Event, error) diff --git a/internal/storage/sqlite.go b/internal/storage/sqlite.go index bc7ce49..2d90eb7 100644 --- a/internal/storage/sqlite.go +++ b/internal/storage/sqlite.go @@ -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 {