From d4285607afb923787eab8ad7d501be1d60a4edf5 Mon Sep 17 00:00:00 2001 From: kami Date: Thu, 30 Jul 2026 23:39:26 +0400 Subject: [PATCH] Persist execution audit fields and enforce the in-flight guard in SQL MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Findings 4 and 7 of REVIEW-2026-07-30.md, as a single user_version step (11 -> 12) because they alter the same table. confirmation_id was written by the INSERT and absent from all three executions SELECTs, so it always read back empty: "which confirmation authorized this destructive act" was unanswerable from the API. causation_id was set on the struct by the engine with no column to land in, and capability_version was missing entirely despite spec §4.1 and the precedent on confirmations. All three are now persisted and selected back. The one-in-flight-per-(capability, target) rule was check-then-insert with no constraint between, so two concurrent requests could both proceed. It is now a partial UNIQUE index; the engine's pre-insert query is demoted to an advisory fast path and a constraint violation maps to ErrExecutionInFlight (409), kept distinct from the idempotency-key replay case. Note the migration also rewrites pre-existing duplicate in-flight rows to status='unknown', keeping the lowest id per pair — creating the index would otherwise fail outright on any database holding stale started rows, which the old non-unique index permitted indefinitely. That is a write to existing audit rows, not just DDL. Rows predating this migration get capability_version=0, which is indistinguishable from a genuine 0; backfill is not possible. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Uea55zaiWuEByEDC4UBSdd --- internal/execution/audit_test.go | 153 +++++++++++++++++++++++++++++++ internal/execution/engine.go | 6 +- internal/storage/sqlite.go | 57 +++++++++--- 3 files changed, 203 insertions(+), 13 deletions(-) create mode 100644 internal/execution/audit_test.go diff --git a/internal/execution/audit_test.go b/internal/execution/audit_test.go new file mode 100644 index 0000000..922a5ad --- /dev/null +++ b/internal/execution/audit_test.go @@ -0,0 +1,153 @@ +package execution_test + +import ( + "errors" + "sync" + "testing" + "time" + + "github.com/kami/hexis/internal/domain" +) + +// The audit trail is only useful if it survives a round trip through storage: +// "which confirmation authorized this destructive act" must be answerable from +// a plain read of the execution, not just from the in-memory struct the engine +// happened to return. +func TestExecute_AuditFieldsRoundTripThroughStorage(t *testing.T) { + prov := &fakeProvider{name: "fake"} + eng, store := newTestEngine(t, prov) + cap := mustCreateCapability(t, store, func(c *domain.Capability) { + c.RequiresConfirmation = true + c.Risk = "destructive" + }) + + 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, + CorrelationID: "corr_1", + CausationID: "caus_1", + IdempotencyKey: "idem_1", + }) + if err != nil { + t.Fatalf("execute: %v", err) + } + + // This is what GET /api/v1/executions/{id} serves. + got, err := store.GetExecution(res.Execution.ID) + if err != nil { + t.Fatalf("get execution: %v", err) + } + if got.ConfirmationID != conf.ID { + t.Errorf("confirmation_id: got %q, want %q", got.ConfirmationID, conf.ID) + } + if got.CausationID != "caus_1" { + t.Errorf("causation_id: got %q, want %q", got.CausationID, "caus_1") + } + if got.CorrelationID != "corr_1" { + t.Errorf("correlation_id: got %q, want %q", got.CorrelationID, "corr_1") + } + if got.CapabilityVersion != cap.Version { + t.Errorf("capability_version: got %d, want %d", got.CapabilityVersion, cap.Version) + } + + // The idempotency-key read path serves the same record and must agree. + byKey, err := store.GetExecutionByIdempotencyKey("idem_1") + if err != nil { + t.Fatalf("get by idempotency key: %v", err) + } + if byKey.ConfirmationID != conf.ID || byKey.CausationID != "caus_1" || byKey.CapabilityVersion != cap.Version { + t.Errorf("idempotency read path lost audit fields: %+v", byKey) + } +} + +// The in-flight guard used to be check-then-insert. Two concurrent identical +// executes must produce exactly one execution row, with the loser reported as +// already-in-flight rather than a 500. +func TestExecute_ConcurrentDuplicateProducesExactlyOneRow(t *testing.T) { + prov := &fakeProvider{name: "fake", delay: 150 * time.Millisecond} + eng, store := newTestEngine(t, prov) + cap := mustCreateCapability(t, store, nil) + + const n = 4 + var wg sync.WaitGroup + errs := make([]error, n) + start := make(chan struct{}) + for i := 0; i < n; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + <-start + _, errs[i] = eng.Execute(&domain.ExecuteRequest{ + CapabilityID: cap.ID, + TargetEntityID: "ent_x", + }) + }(i) + } + close(start) + wg.Wait() + + succeeded := 0 + for i, err := range errs { + switch { + case err == nil: + succeeded++ + case errors.Is(err, domain.ErrExecutionInFlight): + default: + t.Errorf("goroutine %d: unexpected error %v (want nil or ErrExecutionInFlight)", i, err) + } + } + if succeeded != 1 { + t.Errorf("expected exactly 1 successful execute, got %d", succeeded) + } + + var rows int + if err := store.DB().QueryRow( + `SELECT COUNT(*) FROM executions WHERE capability_id = ? AND target_entity_id = ?`, + cap.ID, "ent_x", + ).Scan(&rows); err != nil { + t.Fatalf("count executions: %v", err) + } + if rows != 1 { + t.Errorf("expected exactly 1 execution row, got %d", rows) + } +} + +// Directly exercise the partial unique index, bypassing the engine's advisory +// pre-check, so the constraint itself is what is under test. +func TestStore_SecondInFlightRowRejectedByIndex(t *testing.T) { + _, store := newTestEngine(t, &fakeProvider{name: "fake"}) + now := time.Now().UTC() + + mk := func(status domain.ExecutionStatus) *domain.Execution { + return &domain.Execution{ + ID: domain.NewExecutionID(), + CapabilityID: "cap_1", + TargetEntityID: "ent_x", + Status: status, + CreatedAt: now, + UpdatedAt: now, + } + } + + if err := store.CreateExecution(mk(domain.ExecutionStarted)); err != nil { + t.Fatalf("first insert: %v", err) + } + if err := store.CreateExecution(mk(domain.ExecutionStarted)); !errors.Is(err, domain.ErrExecutionInFlight) { + t.Fatalf("second in-flight insert: got %v, want ErrExecutionInFlight", err) + } + // The index is partial: terminal rows for the same pair stay insertable. + if err := store.CreateExecution(mk(domain.ExecutionSucceeded)); err != nil { + t.Fatalf("terminal-status insert should be allowed: %v", err) + } + if err := store.CreateExecution(mk(domain.ExecutionSucceeded)); err != nil { + t.Fatalf("second terminal-status insert should be allowed: %v", err) + } +} diff --git a/internal/execution/engine.go b/internal/execution/engine.go index 1c9132d..cbc03a4 100644 --- a/internal/execution/engine.go +++ b/internal/execution/engine.go @@ -53,7 +53,10 @@ func (e *Engine) Execute(req *domain.ExecuteRequest) (*ExecuteResult, error) { } } - // One in-flight execution per (capability_id, target_entity_id). + // One in-flight execution per (capability_id, target_entity_id). This check + // is advisory — it gives a clean error before doing any work — but the + // authoritative guard is the partial unique index enforced at INSERT below, + // which closes the check-then-insert window between the two. if _, err := e.store.GetInFlightExecution(req.CapabilityID, req.TargetEntityID); err == nil { return nil, domain.ErrExecutionInFlight } @@ -67,6 +70,7 @@ func (e *Engine) Execute(req *domain.ExecuteRequest) (*ExecuteResult, error) { exec := &domain.Execution{ ID: domain.NewExecutionID(), CapabilityID: req.CapabilityID, + CapabilityVersion: capability.Version, TargetEntityID: req.TargetEntityID, EntityVersion: req.EntityVersion, Arguments: req.Arguments, diff --git a/internal/storage/sqlite.go b/internal/storage/sqlite.go index 2d90eb7..1adfc2d 100644 --- a/internal/storage/sqlite.go +++ b/internal/storage/sqlite.go @@ -141,6 +141,27 @@ var migrations = []string{ state TEXT NOT NULL DEFAULT 'pending' )`, `CREATE INDEX IF NOT EXISTS idx_executions_inflight ON executions(capability_id, target_entity_id, status)`, + // Audit columns missing from executions (spec §4.1) plus a partial UNIQUE + // index enforcing one in-flight execution per (capability, target). + // Pre-existing duplicate in-flight rows would make the index creation fail, + // so they are resolved to 'unknown' first — that is the spec's outcome for + // an execution whose side effect can no longer be determined. + `ALTER TABLE executions ADD COLUMN causation_id TEXT NOT NULL DEFAULT ''; + ALTER TABLE executions ADD COLUMN capability_version INTEGER NOT NULL DEFAULT 0; + UPDATE executions SET status = 'unknown' + WHERE status = 'started' + AND id NOT IN ( + SELECT MIN(id) FROM executions WHERE status = 'started' + GROUP BY capability_id, target_entity_id + ); + DROP INDEX IF EXISTS idx_executions_inflight; + CREATE UNIQUE INDEX IF NOT EXISTS idx_executions_inflight + ON executions(capability_id, target_entity_id) WHERE status = 'started';`, + // Supports the entity_id filter on GET /api/v1/executions (spec §4.5). The + // `since` cursor is the row's implicit rowid, which cannot appear in an + // index (SQLite indexes it as the btree key already), so ordering and + // range-scanning by rowid needs no index of its own. + `CREATE INDEX IF NOT EXISTS idx_executions_entity ON executions(target_entity_id)`, } const timeFmt = "2006-01-02T15:04:05.999999999Z07:00" @@ -292,12 +313,20 @@ 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, 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), + `INSERT INTO executions (id, capability_id, capability_version, target_entity_id, entity_version, arguments, requested_by, origin, idempotency_key, confirmation_id, status, result, error, resolution_evidence, correlation_id, causation_id, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + e.ID, e.CapabilityID, e.CapabilityVersion, 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, e.CausationID, formatTime(e.CreatedAt), formatTime(e.UpdatedAt), ) if err != nil { if strings.Contains(err.Error(), "UNIQUE") { + // The partial unique index on (capability_id, target_entity_id) + // WHERE status='started' is the in-flight guard, not a replay. + if strings.Contains(err.Error(), "idempotency_key") { + return domain.ErrIdempotencyReplay + } + if strings.Contains(err.Error(), "capability_id") || strings.Contains(err.Error(), "target_entity_id") { + return domain.ErrExecutionInFlight + } return domain.ErrIdempotencyReplay } return err @@ -310,12 +339,12 @@ func (s *Store) GetExecution(id string) (*domain.Execution, error) { defer s.mu.RUnlock() 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 + `SELECT id, capability_id, capability_version, target_entity_id, entity_version, arguments, requested_by, origin, COALESCE(idempotency_key,''), COALESCE(confirmation_id,''), status, result, COALESCE(error,''), resolution_evidence, COALESCE(correlation_id,''), COALESCE(causation_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) + var args, reqBy, origin, idempKey, confID, status, result, errStr, evidence, corrID, causID, createdAt, updatedAt string + err := row.Scan(&e.ID, &e.CapabilityID, &e.CapabilityVersion, &e.TargetEntityID, &e.EntityVersion, &args, &reqBy, &origin, &idempKey, &confID, &status, &result, &errStr, &evidence, &corrID, &causID, &createdAt, &updatedAt) if err == sql.ErrNoRows { return nil, domain.ErrExecutionNotFound } @@ -328,9 +357,11 @@ func (s *Store) GetExecution(id string) (*domain.Execution, error) { json.Unmarshal([]byte(result), &e.Result) json.Unmarshal([]byte(evidence), &e.ResolutionEvidence) e.IdempotencyKey = idempKey + e.ConfirmationID = confID e.Status = domain.ExecutionStatus(status) e.Error = errStr e.CorrelationID = corrID + e.CausationID = causID e.CreatedAt = parseTime(createdAt) e.UpdatedAt = parseTime(updatedAt) if e.Arguments == nil { @@ -366,12 +397,12 @@ func (s *Store) GetExecutionByIdempotencyKey(key string) (*domain.Execution, err defer s.mu.RUnlock() row := s.db.QueryRow( - `SELECT 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 + `SELECT id, capability_id, capability_version, target_entity_id, entity_version, arguments, requested_by, origin, COALESCE(idempotency_key,''), COALESCE(confirmation_id,''), status, result, COALESCE(error,''), resolution_evidence, COALESCE(correlation_id,''), COALESCE(causation_id,''), created_at, updated_at FROM executions WHERE idempotency_key = ?`, key, ) 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) + var args, reqBy, origin, idempKey, confID, status, result, errStr, evidence, corrID, causID, createdAt, updatedAt string + err := row.Scan(&e.ID, &e.CapabilityID, &e.CapabilityVersion, &e.TargetEntityID, &e.EntityVersion, &args, &reqBy, &origin, &idempKey, &confID, &status, &result, &errStr, &evidence, &corrID, &causID, &createdAt, &updatedAt) if err == sql.ErrNoRows { return nil, domain.ErrExecutionNotFound } @@ -384,9 +415,11 @@ func (s *Store) GetExecutionByIdempotencyKey(key string) (*domain.Execution, err json.Unmarshal([]byte(result), &e.Result) json.Unmarshal([]byte(evidence), &e.ResolutionEvidence) e.IdempotencyKey = idempKey + e.ConfirmationID = confID e.Status = domain.ExecutionStatus(status) e.Error = errStr e.CorrelationID = corrID + e.CausationID = causID e.CreatedAt = parseTime(createdAt) e.UpdatedAt = parseTime(updatedAt) return e, nil @@ -415,12 +448,12 @@ func (s *Store) GetInFlightExecution(capabilityID, targetEntityID string) (*doma // 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 + `SELECT id, capability_id, capability_version, target_entity_id, entity_version, arguments, requested_by, origin, COALESCE(idempotency_key,''), COALESCE(confirmation_id,''), status, result, COALESCE(error,''), resolution_evidence, COALESCE(correlation_id,''), COALESCE(causation_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) + var args, reqBy, origin, idempKey, confID, status, result, errStr, evidence, corrID, causID, createdAt, updatedAt string + err := row.Scan(&e.ID, &e.CapabilityID, &e.CapabilityVersion, &e.TargetEntityID, &e.EntityVersion, &args, &reqBy, &origin, &idempKey, &confID, &status, &result, &errStr, &evidence, &corrID, &causID, &createdAt, &updatedAt) if err == sql.ErrNoRows { return nil, domain.ErrExecutionNotFound }