diff --git a/adapters/continuityexecutor/executor.go b/adapters/continuityexecutor/executor.go new file mode 100644 index 0000000..51f1dcb --- /dev/null +++ b/adapters/continuityexecutor/executor.go @@ -0,0 +1,96 @@ +package continuityexecutor + +import ( + "bytes" + "encoding/json" + "fmt" + "net/http" + "net/url" + "strings" + + commitcontract "git.cin.su/m/Assistant/contracts/commit" + "git.cin.su/m/Assistant/contracts/continuity" + "git.cin.su/m/Assistant/modules/commit/ports" +) + +type Adapter struct { + baseURL string + token string + client *http.Client +} + +func New(baseURL, token string, client *http.Client) (*Adapter, error) { + baseURL = strings.TrimRight(baseURL, "/") + if baseURL == "" || token == "" || client == nil { + return nil, fmt.Errorf("continuity executor requires base URL, token, and HTTP client") + } + if _, err := url.ParseRequestURI(baseURL); err != nil { + return nil, fmt.Errorf("invalid continuity executor URL: %w", err) + } + return &Adapter{baseURL: baseURL, token: token, client: client}, nil +} + +func (a *Adapter) Execute(action commitcontract.AuthorizedAction) ports.ExecutionResult { + if action.Proposal.CapabilityID != commitcontract.CapabilityContinuityTransition || action.Proposal.Operation != commitcontract.OperationSatisfyDelivered { + return ports.ExecutionResult{Known: true, Succeeded: false} + } + command := continuity.SatisfyDeliveredCommand{ + ActionIntentID: continuity.CommittedActionIntentID(action.Proposal.IntentID), + AuthorizedActionID: continuity.CommittedAuthorizedActionID(action.ID), + OwnerPrincipalID: action.PrincipalID, + Subject: action.Proposal.ContinuitySubject, + } + body, err := json.Marshal(command) + if err != nil { + return ports.ExecutionResult{Known: true, Succeeded: false} + } + request, err := http.NewRequest(http.MethodPost, a.baseURL+"/v1/internal/executor/satisfy-delivered", bytes.NewReader(body)) + if err != nil { + return ports.ExecutionResult{Known: true, Succeeded: false} + } + request.Header.Set("Authorization", "Bearer "+a.token) + request.Header.Set("Content-Type", "application/json") + response, err := a.client.Do(request) + if err != nil { + return ports.ExecutionResult{Known: false} + } + defer response.Body.Close() + if response.StatusCode == http.StatusOK { + var receipt continuity.ActionReceipt + if json.NewDecoder(response.Body).Decode(&receipt) != nil || + receipt.ActionIntentID != command.ActionIntentID || receipt.AuthorizedActionID != command.AuthorizedActionID || + receipt.ResultState != continuity.ProspectiveSatisfied { + return ports.ExecutionResult{Known: false} + } + return ports.ExecutionResult{Known: true, Succeeded: true} + } + if response.StatusCode >= 400 && response.StatusCode < 500 { + return ports.ExecutionResult{Known: true, Succeeded: false} + } + return ports.ExecutionResult{Known: false} +} + +func (a *Adapter) Reconcile(action commitcontract.AuthorizedAction) ports.ExecutionResult { + endpoint := a.baseURL + "/v1/internal/executor/receipts/" + url.PathEscape(string(action.Proposal.IntentID)) + request, err := http.NewRequest(http.MethodGet, endpoint, nil) + if err != nil { + return ports.ExecutionResult{Known: false} + } + request.Header.Set("Authorization", "Bearer "+a.token) + response, err := a.client.Do(request) + if err != nil { + return ports.ExecutionResult{Known: false} + } + defer response.Body.Close() + if response.StatusCode != http.StatusOK { + return ports.ExecutionResult{Known: false} + } + var receipt continuity.ActionReceipt + if json.NewDecoder(response.Body).Decode(&receipt) != nil || + receipt.ActionIntentID != continuity.CommittedActionIntentID(action.Proposal.IntentID) || + receipt.AuthorizedActionID != continuity.CommittedAuthorizedActionID(action.ID) || + receipt.OwnerPrincipalID != action.PrincipalID || receipt.Subject != action.Proposal.ContinuitySubject { + return ports.ExecutionResult{Known: false} + } + return ports.ExecutionResult{Known: true, Succeeded: receipt.ResultState == continuity.ProspectiveSatisfied} +} diff --git a/adapters/continuityexecutor/executor_test.go b/adapters/continuityexecutor/executor_test.go new file mode 100644 index 0000000..5c801da --- /dev/null +++ b/adapters/continuityexecutor/executor_test.go @@ -0,0 +1,89 @@ +package continuityexecutor + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + commitcontract "git.cin.su/m/Assistant/contracts/commit" + "git.cin.su/m/Assistant/contracts/continuity" +) + +func TestExecuteAndReconcileUseTypedReceipt(t *testing.T) { + action := testAction() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Authorization") != "Bearer secret" { + http.Error(w, "denied", http.StatusUnauthorized) + return + } + receipt := continuity.ActionReceipt{ + ActionIntentID: continuity.CommittedActionIntentID(action.Proposal.IntentID), AuthorizedActionID: continuity.CommittedAuthorizedActionID(action.ID), + OwnerPrincipalID: action.PrincipalID, Subject: action.Proposal.ContinuitySubject, + ResultState: continuity.ProspectiveSatisfied, ResultRevision: 5, + } + _ = json.NewEncoder(w).Encode(receipt) + })) + defer server.Close() + adapter, err := New(server.URL, "secret", server.Client()) + if err != nil { + t.Fatal(err) + } + if result := adapter.Execute(action); !result.Known || !result.Succeeded { + t.Fatalf("execute result = %#v", result) + } + if result := adapter.Reconcile(action); !result.Known || !result.Succeeded { + t.Fatalf("reconcile result = %#v", result) + } +} + +func TestAcknowledgementLossReconcilesWithoutDuplicatePost(t *testing.T) { + action := testAction() + posts := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodPost { + posts++ + hijacker, ok := w.(http.Hijacker) + if !ok { + t.Fatal("test server cannot simulate acknowledgement loss") + } + connection, _, err := hijacker.Hijack() + if err != nil { + t.Fatal(err) + } + _ = connection.Close() + return + } + receipt := continuity.ActionReceipt{ + ActionIntentID: continuity.CommittedActionIntentID(action.Proposal.IntentID), AuthorizedActionID: continuity.CommittedAuthorizedActionID(action.ID), + OwnerPrincipalID: action.PrincipalID, Subject: action.Proposal.ContinuitySubject, + ResultState: continuity.ProspectiveSatisfied, ResultRevision: 5, + } + _ = json.NewEncoder(w).Encode(receipt) + })) + defer server.Close() + adapter, err := New(server.URL, "secret", server.Client()) + if err != nil { + t.Fatal(err) + } + if result := adapter.Execute(action); result.Known { + t.Fatalf("lost acknowledgement was treated as known: %#v", result) + } + if result := adapter.Reconcile(action); !result.Known || !result.Succeeded { + t.Fatalf("reconciliation result = %#v", result) + } + if posts != 1 { + t.Fatalf("effect was posted %d times", posts) + } +} + +func testAction() commitcontract.AuthorizedAction { + return commitcontract.AuthorizedAction{ + ID: "authorized-1", PrincipalID: "principal-1", State: commitcontract.ActionAuthorized, + Proposal: commitcontract.ActionProposal{ + IntentID: "intent-1", CapabilityID: commitcontract.CapabilityContinuityTransition, + Operation: commitcontract.OperationSatisfyDelivered, + ContinuitySubject: continuity.SatisfyDeliveredSubject{ProspectiveItemID: "prospective-1", ExpectedRevision: 4, FromState: continuity.ProspectiveDelivered, ToState: continuity.ProspectiveSatisfied, EvidenceRef: "artifact://evidence/1"}, + }, + } +} diff --git a/adapters/postgres/commit_repository.go b/adapters/postgres/commit_repository.go index bde2a33..d4c2ee8 100644 --- a/adapters/postgres/commit_repository.go +++ b/adapters/postgres/commit_repository.go @@ -2,11 +2,22 @@ package postgres import ( "context" + "errors" "fmt" + "reflect" + commitcontract "git.cin.su/m/Assistant/contracts/commit" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" ) +var ( + ErrAuthorizedActionNotFound = errors.New("authorized action not found") + ErrAuthorizedActionConflict = errors.New("authorized action identity conflict") + ErrActionAttemptNotFound = errors.New("action attempt not found") + ErrActionAttemptConflict = errors.New("action attempt identity conflict") +) + type CommitRepository struct{ pool *pgxpool.Pool } func OpenCommit(ctx context.Context, c ContinuityConfig) (*CommitRepository, error) { @@ -18,10 +29,112 @@ func OpenCommit(ctx context.Context, c ContinuityConfig) (*CommitRepository, err } func (r *CommitRepository) Close() { r.pool.Close() } func (r *CommitRepository) SaveAuthorized(ctx context.Context, a commitcontract.AuthorizedAction) error { - _, e := r.pool.Exec(ctx, `INSERT INTO commit.authorized_actions (authorized_action_id,action_intent_id,state,revision) VALUES ($1,$2,$3,1) ON CONFLICT (authorized_action_id) DO NOTHING`, a.ID, a.Proposal.IntentID, a.State) - return e + tx, err := r.pool.Begin(ctx) + if err != nil { + return err + } + defer func() { _ = tx.Rollback(ctx) }() + if _, err = tx.Exec(ctx, `INSERT INTO commit.authorized_actions (authorized_action_id,action_intent_id,state,revision) VALUES ($1,$2,$3,1) ON CONFLICT (authorized_action_id) DO NOTHING`, a.ID, a.Proposal.IntentID, a.State); err != nil { + return err + } + result, err := tx.Exec(ctx, `INSERT INTO commit.authorized_action_envelopes ( + authorized_action_id,action_intent_id,actor_id,principal_id,grant_id,grant_revision,scope_id, + task_id,goal_id,task_scope_id,task_revision,capability_id,operation,prospective_item_id, + expected_prospective_revision,prospective_from_state,prospective_to_state,evidence_ref, + proposal_state,action_state,authorized_at_unix_millis,envelope_revision) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,1) + ON CONFLICT DO NOTHING`, + a.ID, a.Proposal.IntentID, a.Proposal.ActorID, a.PrincipalID, a.GrantID, a.GrantRevision, a.ScopeID, + a.Proposal.Target.TaskID, a.Proposal.Target.GoalID, a.Proposal.Target.ScopeID, a.Proposal.Target.Revision, + a.Proposal.CapabilityID, a.Proposal.Operation, a.Proposal.ContinuitySubject.ProspectiveItemID, + a.Proposal.ContinuitySubject.ExpectedRevision, a.Proposal.ContinuitySubject.FromState, + a.Proposal.ContinuitySubject.ToState, a.Proposal.ContinuitySubject.EvidenceRef, + a.Proposal.State, a.State, a.AuthorizedAtUnixMillis) + if err != nil { + return err + } + if result.RowsAffected() == 0 { + existing, loadErr := loadAuthorized(ctx, tx, a.ID) + if loadErr != nil || !reflect.DeepEqual(existing, a) { + return ErrAuthorizedActionConflict + } + } + return tx.Commit(ctx) } func (r *CommitRepository) SaveAttempt(ctx context.Context, a commitcontract.ActionAttempt) error { - _, e := r.pool.Exec(ctx, `INSERT INTO commit.action_attempts (action_attempt_id,authorized_action_id,action_intent_id,state) VALUES ($1,$2,$3,$4) ON CONFLICT (action_attempt_id) DO NOTHING`, a.ID, a.AuthorizedID, a.IntentID, a.State) - return e + result, err := r.pool.Exec(ctx, `INSERT INTO commit.action_attempts (action_attempt_id,authorized_action_id,action_intent_id,state) VALUES ($1,$2,$3,$4) ON CONFLICT (action_attempt_id) DO NOTHING`, a.ID, a.AuthorizedID, a.IntentID, a.State) + if err != nil { + return err + } + if result.RowsAffected() == 0 { + var existing commitcontract.ActionAttempt + err = r.pool.QueryRow(ctx, `SELECT action_attempt_id,authorized_action_id,action_intent_id,state FROM commit.action_attempts WHERE action_attempt_id=$1`, a.ID).Scan(&existing.ID, &existing.AuthorizedID, &existing.IntentID, &existing.State) + if err != nil || existing != a { + return ErrActionAttemptConflict + } + } + return nil +} + +func (r *CommitRepository) BeginAttempt(ctx context.Context, a commitcontract.ActionAttempt) error { + if a.State != commitcontract.ActionExecuting { + return ErrActionAttemptConflict + } + return r.SaveAttempt(ctx, a) +} + +func (r *CommitRepository) FinishAttempt(ctx context.Context, a commitcontract.ActionAttempt) error { + if a.State != commitcontract.ActionSucceeded && a.State != commitcontract.ActionFailed && a.State != commitcontract.ActionUnknownOutcome { + return ErrActionAttemptConflict + } + result, err := r.pool.Exec(ctx, `UPDATE commit.action_attempts SET state=$1,revision=revision+1,updated_at=CURRENT_TIMESTAMP + WHERE action_attempt_id=$2 AND authorized_action_id=$3 AND action_intent_id=$4 AND state=$5`, + a.State, a.ID, a.AuthorizedID, a.IntentID, commitcontract.ActionExecuting) + if err != nil { + return err + } + if result.RowsAffected() != 1 { + return ErrActionAttemptConflict + } + return nil +} + +func (r *CommitRepository) LoadAuthorized(ctx context.Context, id commitcontract.AuthorizedActionID) (commitcontract.AuthorizedAction, error) { + return loadAuthorized(ctx, r.pool, id) +} + +func (r *CommitRepository) LoadLatestAttempt(ctx context.Context, id commitcontract.AuthorizedActionID) (commitcontract.ActionAttempt, error) { + var attempt commitcontract.ActionAttempt + err := r.pool.QueryRow(ctx, `SELECT action_attempt_id,authorized_action_id,action_intent_id,state FROM commit.action_attempts WHERE authorized_action_id=$1 ORDER BY recorded_at DESC, action_attempt_id DESC LIMIT 1`, id).Scan(&attempt.ID, &attempt.AuthorizedID, &attempt.IntentID, &attempt.State) + if errors.Is(err, pgx.ErrNoRows) { + return commitcontract.ActionAttempt{}, ErrActionAttemptNotFound + } + return attempt, err +} + +type commitRowQuerier interface { + QueryRow(context.Context, string, ...any) pgx.Row +} + +func loadAuthorized(ctx context.Context, querier commitRowQuerier, id commitcontract.AuthorizedActionID) (commitcontract.AuthorizedAction, error) { + var action commitcontract.AuthorizedAction + err := querier.QueryRow(ctx, `SELECT authorized_action_id,action_intent_id,actor_id,principal_id,grant_id,grant_revision,scope_id, + task_id,goal_id,task_scope_id,task_revision,capability_id,operation,prospective_item_id, + expected_prospective_revision,prospective_from_state,prospective_to_state,evidence_ref, + proposal_state,action_state,authorized_at_unix_millis + FROM commit.authorized_action_envelopes WHERE authorized_action_id=$1`, id).Scan( + &action.ID, &action.Proposal.IntentID, &action.Proposal.ActorID, &action.PrincipalID, + &action.GrantID, &action.GrantRevision, &action.ScopeID, &action.Proposal.Target.TaskID, + &action.Proposal.Target.GoalID, &action.Proposal.Target.ScopeID, &action.Proposal.Target.Revision, + &action.Proposal.CapabilityID, &action.Proposal.Operation, &action.Proposal.ContinuitySubject.ProspectiveItemID, + &action.Proposal.ContinuitySubject.ExpectedRevision, &action.Proposal.ContinuitySubject.FromState, + &action.Proposal.ContinuitySubject.ToState, &action.Proposal.ContinuitySubject.EvidenceRef, + &action.Proposal.State, &action.State, &action.AuthorizedAtUnixMillis) + if errors.Is(err, pgx.ErrNoRows) { + return commitcontract.AuthorizedAction{}, ErrAuthorizedActionNotFound + } + if err != nil { + return commitcontract.AuthorizedAction{}, fmt.Errorf("load authorized action: %w", err) + } + return action, nil } diff --git a/adapters/postgres/continuity_repository.go b/adapters/postgres/continuity_repository.go index 5292324..b3d26b7 100644 --- a/adapters/postgres/continuity_repository.go +++ b/adapters/postgres/continuity_repository.go @@ -7,10 +7,19 @@ import ( "git.cin.su/m/Assistant/contracts/continuity" "git.cin.su/m/Assistant/contracts/identity" + continuitydomain "git.cin.su/m/Assistant/modules/continuity/domain" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" ) -var ErrProspectiveNotFound = errors.New("prospective item not found") +var ( + ErrProspectiveNotFound = errors.New("prospective item not found") + ErrProspectiveOwnerConflict = errors.New("prospective item owner conflict") + ErrProspectiveStateConflict = errors.New("prospective item state conflict") + ErrProspectiveRevisionConflict = errors.New("prospective item revision conflict") + ErrActionReceiptNotFound = errors.New("continuity action receipt not found") + ErrActionIntentConflict = errors.New("action intent was already used for a different command") +) type ContinuityConfig struct{ Host, Port, Database, User, Password string } type ContinuityRepository struct{ pool *pgxpool.Pool } @@ -56,3 +65,117 @@ func (r *ContinuityRepository) ByOwner(ctx context.Context, owner identity.Princ } return items, nil } + +// SatisfyDelivered owns both the Prospective CAS and its idempotency receipt in +// one Continuity-local transaction. It never reaches Commit storage. +func (r *ContinuityRepository) SatisfyDelivered(ctx context.Context, command continuity.SatisfyDeliveredCommand) (continuity.ActionReceipt, error) { + if err := continuitydomain.ValidateSatisfyDelivered(command); err != nil { + return continuity.ActionReceipt{}, err + } + tx, err := r.pool.Begin(ctx) + if err != nil { + return continuity.ActionReceipt{}, err + } + defer func() { _ = tx.Rollback(ctx) }() + // Serialize deliveries of one ActionIntent inside the canonical owner. The + // receipt remains the durable correctness source; the lock is transaction-local. + if _, err = tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,0))`, command.ActionIntentID); err != nil { + return continuity.ActionReceipt{}, err + } + if existing, loadErr := loadActionReceipt(ctx, tx, command.ActionIntentID); loadErr == nil { + if receiptMatchesCommand(existing, command) { + return existing, tx.Commit(ctx) + } + return continuity.ActionReceipt{}, ErrActionIntentConflict + } else if !errors.Is(loadErr, ErrActionReceiptNotFound) { + return continuity.ActionReceipt{}, loadErr + } + + var item continuity.ProspectiveItem + err = tx.QueryRow(ctx, `UPDATE continuity.prospective_items + SET state=$1, revision=revision+1 + WHERE prospective_item_id=$2 AND owner_principal_id=$3 AND state=$4 AND revision=$5 + RETURNING prospective_item_id,commitment_id,owner_principal_id,scope_id,goal_id,trigger_at_unix_millis,state,revision`, + continuity.ProspectiveSatisfied, command.Subject.ProspectiveItemID, command.OwnerPrincipalID, + continuity.ProspectiveDelivered, command.Subject.ExpectedRevision).Scan( + &item.ID, &item.CommitmentID, &item.OwnerPrincipalID, &item.ScopeID, &item.GoalID, + &item.TriggerAtUnixMillis, &item.State, &item.Revision) + if errors.Is(err, pgx.ErrNoRows) { + return continuity.ActionReceipt{}, diagnoseProspectiveConflict(ctx, tx, command) + } + if err != nil { + return continuity.ActionReceipt{}, fmt.Errorf("satisfy delivered prospective item: %w", err) + } + receipt := continuity.ActionReceipt{ + ActionIntentID: command.ActionIntentID, AuthorizedActionID: command.AuthorizedActionID, + OwnerPrincipalID: command.OwnerPrincipalID, Subject: command.Subject, + ResultState: item.State, ResultRevision: item.Revision, + } + _, err = tx.Exec(ctx, `INSERT INTO continuity.action_receipts ( + action_intent_id,authorized_action_id,owner_principal_id,prospective_item_id,expected_revision, + from_state,to_state,evidence_ref,result_state,result_revision) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)`, receipt.ActionIntentID, receipt.AuthorizedActionID, + receipt.OwnerPrincipalID, receipt.Subject.ProspectiveItemID, receipt.Subject.ExpectedRevision, + receipt.Subject.FromState, receipt.Subject.ToState, receipt.Subject.EvidenceRef, + receipt.ResultState, receipt.ResultRevision) + if err != nil { + return continuity.ActionReceipt{}, fmt.Errorf("save continuity action receipt: %w", err) + } + if err = tx.Commit(ctx); err != nil { + return continuity.ActionReceipt{}, err + } + return receipt, nil +} + +func (r *ContinuityRepository) LoadActionReceipt(ctx context.Context, intentID continuity.CommittedActionIntentID) (continuity.ActionReceipt, error) { + return loadActionReceipt(ctx, r.pool, intentID) +} + +type continuityRowQuerier interface { + QueryRow(context.Context, string, ...any) pgx.Row +} + +func loadActionReceipt(ctx context.Context, querier continuityRowQuerier, intentID continuity.CommittedActionIntentID) (continuity.ActionReceipt, error) { + var receipt continuity.ActionReceipt + err := querier.QueryRow(ctx, `SELECT action_intent_id,authorized_action_id,owner_principal_id,prospective_item_id, + expected_revision,from_state,to_state,evidence_ref,result_state,result_revision + FROM continuity.action_receipts WHERE action_intent_id=$1`, intentID).Scan( + &receipt.ActionIntentID, &receipt.AuthorizedActionID, &receipt.OwnerPrincipalID, + &receipt.Subject.ProspectiveItemID, &receipt.Subject.ExpectedRevision, &receipt.Subject.FromState, + &receipt.Subject.ToState, &receipt.Subject.EvidenceRef, &receipt.ResultState, &receipt.ResultRevision) + if errors.Is(err, pgx.ErrNoRows) { + return continuity.ActionReceipt{}, ErrActionReceiptNotFound + } + if err != nil { + return continuity.ActionReceipt{}, fmt.Errorf("load continuity action receipt: %w", err) + } + return receipt, nil +} + +func receiptMatchesCommand(receipt continuity.ActionReceipt, command continuity.SatisfyDeliveredCommand) bool { + return receipt.ActionIntentID == command.ActionIntentID && receipt.AuthorizedActionID == command.AuthorizedActionID && + receipt.OwnerPrincipalID == command.OwnerPrincipalID && receipt.Subject == command.Subject +} + +func diagnoseProspectiveConflict(ctx context.Context, querier continuityRowQuerier, command continuity.SatisfyDeliveredCommand) error { + var owner identity.PrincipalID + var state continuity.ProspectiveState + var revision continuity.ProspectiveRevision + err := querier.QueryRow(ctx, `SELECT owner_principal_id,state,revision FROM continuity.prospective_items WHERE prospective_item_id=$1`, command.Subject.ProspectiveItemID).Scan(&owner, &state, &revision) + if errors.Is(err, pgx.ErrNoRows) { + return ErrProspectiveNotFound + } + if err != nil { + return err + } + if owner != command.OwnerPrincipalID { + return ErrProspectiveOwnerConflict + } + if state != continuity.ProspectiveDelivered { + return ErrProspectiveStateConflict + } + if revision != command.Subject.ExpectedRevision { + return ErrProspectiveRevisionConflict + } + return ErrProspectiveRevisionConflict +} diff --git a/cmd/commit-runtime/execution.go b/cmd/commit-runtime/execution.go new file mode 100644 index 0000000..ee1d5c6 --- /dev/null +++ b/cmd/commit-runtime/execution.go @@ -0,0 +1,133 @@ +package main + +import ( + "context" + "crypto/subtle" + "encoding/json" + "errors" + "net/http" + "strings" + + "git.cin.su/m/Assistant/adapters/postgres" + commitcontract "git.cin.su/m/Assistant/contracts/commit" + commitdomain "git.cin.su/m/Assistant/modules/commit/domain" + "git.cin.su/m/Assistant/modules/commit/ports" +) + +type executionRepository interface { + LoadAuthorized(context.Context, commitcontract.AuthorizedActionID) (commitcontract.AuthorizedAction, error) + LoadLatestAttempt(context.Context, commitcontract.AuthorizedActionID) (commitcontract.ActionAttempt, error) + BeginAttempt(context.Context, commitcontract.ActionAttempt) error + FinishAttempt(context.Context, commitcontract.ActionAttempt) error +} + +type attemptRequest struct { + AttemptID commitcontract.ActionAttemptID +} + +func registerExecutionRoutes(mux *http.ServeMux, repository executionRepository, executor ports.Executor, reconciler ports.Reconciler, token string) { + mux.HandleFunc("POST /v1/internal/authorized-actions/{id}/execute", requireCommitToken(token, func(w http.ResponseWriter, r *http.Request) { + action, err := repository.LoadAuthorized(r.Context(), commitcontract.AuthorizedActionID(r.PathValue("id"))) + if errors.Is(err, postgres.ErrAuthorizedActionNotFound) { + http.Error(w, "authorized action not found", http.StatusNotFound) + return + } + if err != nil { + http.Error(w, "authorized action unavailable", http.StatusInternalServerError) + return + } + if _, err = repository.LoadLatestAttempt(r.Context(), action.ID); err == nil { + http.Error(w, "action already has an execution attempt", http.StatusConflict) + return + } else if !errors.Is(err, postgres.ErrActionAttemptNotFound) { + http.Error(w, "action attempt unavailable", http.StatusInternalServerError) + return + } + request, ok := decodeAttemptRequest(w, r) + if !ok { + return + } + begun := commitcontract.ActionAttempt{ID: request.AttemptID, AuthorizedID: action.ID, IntentID: action.Proposal.IntentID, State: commitcontract.ActionExecuting} + if err = repository.BeginAttempt(r.Context(), begun); err != nil { + http.Error(w, "begin execution attempt", http.StatusConflict) + return + } + attempt, executeErr := commitdomain.Execute(action, request.AttemptID, executor) + if err = repository.FinishAttempt(r.Context(), attempt); err != nil { + http.Error(w, "finish execution attempt", http.StatusInternalServerError) + return + } + writeAttempt(w, attempt, executeErr) + })) + + mux.HandleFunc("POST /v1/internal/authorized-actions/{id}/reconcile", requireCommitToken(token, func(w http.ResponseWriter, r *http.Request) { + action, err := repository.LoadAuthorized(r.Context(), commitcontract.AuthorizedActionID(r.PathValue("id"))) + if errors.Is(err, postgres.ErrAuthorizedActionNotFound) { + http.Error(w, "authorized action not found", http.StatusNotFound) + return + } + if err != nil { + http.Error(w, "authorized action unavailable", http.StatusInternalServerError) + return + } + previous, err := repository.LoadLatestAttempt(r.Context(), action.ID) + if err != nil { + http.Error(w, "unknown execution attempt not found", http.StatusConflict) + return + } + if previous.State == commitcontract.ActionExecuting { + previous.State = commitcontract.ActionUnknownOutcome + if err = repository.FinishAttempt(r.Context(), previous); err != nil { + http.Error(w, "recover interrupted attempt", http.StatusInternalServerError) + return + } + } + if previous.State != commitcontract.ActionUnknownOutcome { + http.Error(w, "action does not require reconciliation", http.StatusConflict) + return + } + request, ok := decodeAttemptRequest(w, r) + if !ok { + return + } + begun := commitcontract.ActionAttempt{ID: request.AttemptID, AuthorizedID: action.ID, IntentID: action.Proposal.IntentID, State: commitcontract.ActionExecuting} + if err = repository.BeginAttempt(r.Context(), begun); err != nil { + http.Error(w, "begin reconciliation attempt", http.StatusConflict) + return + } + attempt, reconcileErr := commitdomain.Reconcile(action, previous, request.AttemptID, reconciler) + if err = repository.FinishAttempt(r.Context(), attempt); err != nil { + http.Error(w, "finish reconciliation attempt", http.StatusInternalServerError) + return + } + writeAttempt(w, attempt, reconcileErr) + })) +} + +func decodeAttemptRequest(w http.ResponseWriter, r *http.Request) (attemptRequest, bool) { + var request attemptRequest + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 8<<10)).Decode(&request); err != nil || request.AttemptID == "" { + http.Error(w, "invalid attempt request", http.StatusBadRequest) + return attemptRequest{}, false + } + return request, true +} + +func writeAttempt(w http.ResponseWriter, attempt commitcontract.ActionAttempt, operationErr error) { + w.Header().Set("Content-Type", "application/json") + if errors.Is(operationErr, commitdomain.ErrUnknownOutcome) { + w.WriteHeader(http.StatusAccepted) + } + _ = json.NewEncoder(w).Encode(attempt) +} + +func requireCommitToken(expected string, next http.HandlerFunc) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + provided := strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ") + if expected == "" || len(provided) != len(expected) || subtle.ConstantTimeCompare([]byte(provided), []byte(expected)) != 1 { + http.Error(w, "commit executor denied", http.StatusUnauthorized) + return + } + next(w, r) + } +} diff --git a/cmd/commit-runtime/execution_test.go b/cmd/commit-runtime/execution_test.go new file mode 100644 index 0000000..5772222 --- /dev/null +++ b/cmd/commit-runtime/execution_test.go @@ -0,0 +1,73 @@ +package main + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "git.cin.su/m/Assistant/adapters/postgres" + commitcontract "git.cin.su/m/Assistant/contracts/commit" + "git.cin.su/m/Assistant/modules/commit/ports" +) + +type fakeExecutionRepository struct { + action commitcontract.AuthorizedAction + attempts []commitcontract.ActionAttempt +} + +func (f *fakeExecutionRepository) LoadAuthorized(context.Context, commitcontract.AuthorizedActionID) (commitcontract.AuthorizedAction, error) { + return f.action, nil +} +func (f *fakeExecutionRepository) LoadLatestAttempt(context.Context, commitcontract.AuthorizedActionID) (commitcontract.ActionAttempt, error) { + if len(f.attempts) == 0 { + return commitcontract.ActionAttempt{}, postgres.ErrActionAttemptNotFound + } + return f.attempts[len(f.attempts)-1], nil +} +func (f *fakeExecutionRepository) BeginAttempt(_ context.Context, attempt commitcontract.ActionAttempt) error { + f.attempts = append(f.attempts, attempt) + return nil +} +func (f *fakeExecutionRepository) FinishAttempt(_ context.Context, attempt commitcontract.ActionAttempt) error { + if len(f.attempts) == 0 || f.attempts[len(f.attempts)-1].ID != attempt.ID { + return errors.New("attempt not begun") + } + f.attempts[len(f.attempts)-1] = attempt + return nil +} + +type knownExecutor struct{ calls int } + +func (e *knownExecutor) Execute(commitcontract.AuthorizedAction) ports.ExecutionResult { + e.calls++ + return ports.ExecutionResult{Known: true, Succeeded: true} +} +func (e *knownExecutor) Reconcile(commitcontract.AuthorizedAction) ports.ExecutionResult { + return ports.ExecutionResult{Known: true, Succeeded: true} +} + +func TestExecutionIsDurablyBegunAndCannotBeBlindlyRepeated(t *testing.T) { + repository := &fakeExecutionRepository{action: commitcontract.AuthorizedAction{ID: "authorized-1", State: commitcontract.ActionAuthorized, Proposal: commitcontract.ActionProposal{IntentID: "intent-1"}}} + executor := &knownExecutor{} + mux := http.NewServeMux() + registerExecutionRoutes(mux, repository, executor, executor, "secret") + + request := httptest.NewRequest(http.MethodPost, "/v1/internal/authorized-actions/authorized-1/execute", strings.NewReader(`{"AttemptID":"attempt-1"}`)) + request.Header.Set("Authorization", "Bearer secret") + result := httptest.NewRecorder() + mux.ServeHTTP(result, request) + if result.Code != http.StatusOK || executor.calls != 1 || repository.attempts[0].State != commitcontract.ActionSucceeded { + t.Fatalf("status=%d calls=%d attempts=%#v", result.Code, executor.calls, repository.attempts) + } + + repeated := httptest.NewRequest(http.MethodPost, "/v1/internal/authorized-actions/authorized-1/execute", strings.NewReader(`{"AttemptID":"attempt-2"}`)) + repeated.Header.Set("Authorization", "Bearer secret") + repeatedResult := httptest.NewRecorder() + mux.ServeHTTP(repeatedResult, repeated) + if repeatedResult.Code != http.StatusConflict || executor.calls != 1 { + t.Fatalf("repeat status=%d calls=%d", repeatedResult.Code, executor.calls) + } +} diff --git a/cmd/commit-runtime/main.go b/cmd/commit-runtime/main.go index 5658097..ddc7612 100644 --- a/cmd/commit-runtime/main.go +++ b/cmd/commit-runtime/main.go @@ -8,7 +8,9 @@ import ( "net/http" "net/url" "os" + "time" + "git.cin.su/m/Assistant/adapters/continuityexecutor" "git.cin.su/m/Assistant/adapters/postgres" "git.cin.su/m/Assistant/contracts/authority" commitcontract "git.cin.su/m/Assistant/contracts/commit" @@ -52,8 +54,17 @@ func main() { if coreURL == "" { log.Fatal("CORE_RUNTIME_URL is required") } + executorToken := os.Getenv("COMMIT_EXECUTOR_TOKEN") + if executorToken == "" { + log.Fatal("COMMIT_EXECUTOR_TOKEN is required") + } + executor, err := continuityexecutor.New(envOr("CONTINUITY_RUNTIME_URL", "http://continuity-runtime:8081"), executorToken, &http.Client{Timeout: 5 * time.Second}) + if err != nil { + log.Fatal(err) + } mux := http.NewServeMux() mux.Handle("/healthz", health.Handler{Runtime: "commit-runtime", SemanticGeneration: 1, ArchitectureGeneration: 1}) + registerExecutionRoutes(mux, repository, executor, executor, executorToken) mux.HandleFunc("POST /v1/authorized-actions", func(w http.ResponseWriter, r *http.Request) { var request authorizationRequest if err := json.NewDecoder(r.Body).Decode(&request); err != nil { @@ -80,7 +91,7 @@ func main() { http.Error(w, "work target is stale or unavailable", http.StatusConflict) return } - action, err := commitdomain.Authorize(request.Proposal, commitdomain.AuthorizationContext{Binding: binding, Scope: scope, Grant: grant}, authorityValidator{}, request.AuthorizedID) + action, err := commitdomain.Authorize(request.Proposal, commitdomain.AuthorizationContext{Binding: binding, Scope: scope, Grant: grant}, authorityValidator{}, request.AuthorizedID, time.Now().UnixMilli()) if err != nil { http.Error(w, "authorization denied", http.StatusConflict) return @@ -113,6 +124,13 @@ func main() { } } +func envOr(name, fallback string) string { + if value := os.Getenv(name); value != "" { + return value + } + return fallback +} + func loadJSON[T any](ctx context.Context, endpoint string) (T, error) { var value T request, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil) diff --git a/cmd/continuity-runtime/internal_executor.go b/cmd/continuity-runtime/internal_executor.go new file mode 100644 index 0000000..19f5960 --- /dev/null +++ b/cmd/continuity-runtime/internal_executor.go @@ -0,0 +1,69 @@ +package main + +import ( + "context" + "crypto/subtle" + "encoding/json" + "errors" + "net/http" + "strings" + + "git.cin.su/m/Assistant/adapters/postgres" + "git.cin.su/m/Assistant/contracts/continuity" + continuitydomain "git.cin.su/m/Assistant/modules/continuity/domain" +) + +type continuityExecutorRepository interface { + SatisfyDelivered(context.Context, continuity.SatisfyDeliveredCommand) (continuity.ActionReceipt, error) + LoadActionReceipt(context.Context, continuity.CommittedActionIntentID) (continuity.ActionReceipt, error) +} + +func registerInternalExecutorRoutes(mux *http.ServeMux, repository continuityExecutorRepository, token string) { + mux.HandleFunc("POST /v1/internal/executor/satisfy-delivered", requireExecutorToken(token, func(w http.ResponseWriter, r *http.Request) { + var command continuity.SatisfyDeliveredCommand + decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, 32<<10)) + if err := decoder.Decode(&command); err != nil { + http.Error(w, "invalid executor command", http.StatusBadRequest) + return + } + receipt, err := repository.SatisfyDelivered(r.Context(), command) + switch { + case err == nil: + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(receipt) + case errors.Is(err, continuitydomain.ErrInvalidSatisfaction): + http.Error(w, "invalid executor command", http.StatusBadRequest) + case errors.Is(err, postgres.ErrProspectiveNotFound): + http.Error(w, "prospective item not found", http.StatusNotFound) + case errors.Is(err, postgres.ErrProspectiveOwnerConflict), errors.Is(err, postgres.ErrProspectiveStateConflict), + errors.Is(err, postgres.ErrProspectiveRevisionConflict), errors.Is(err, postgres.ErrActionIntentConflict): + http.Error(w, "prospective command conflict", http.StatusConflict) + default: + http.Error(w, "continuity execution unavailable", http.StatusInternalServerError) + } + })) + mux.HandleFunc("GET /v1/internal/executor/receipts/{intentID}", requireExecutorToken(token, func(w http.ResponseWriter, r *http.Request) { + receipt, err := repository.LoadActionReceipt(r.Context(), continuity.CommittedActionIntentID(r.PathValue("intentID"))) + if errors.Is(err, postgres.ErrActionReceiptNotFound) { + http.Error(w, "receipt not found", http.StatusNotFound) + return + } + if err != nil { + http.Error(w, "continuity receipt unavailable", http.StatusInternalServerError) + return + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(receipt) + })) +} + +func requireExecutorToken(expected string, next http.HandlerFunc) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + provided := strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ") + if expected == "" || len(provided) != len(expected) || subtle.ConstantTimeCompare([]byte(provided), []byte(expected)) != 1 { + http.Error(w, "executor denied", http.StatusUnauthorized) + return + } + next(w, r) + } +} diff --git a/cmd/continuity-runtime/internal_executor_test.go b/cmd/continuity-runtime/internal_executor_test.go new file mode 100644 index 0000000..78eb6ed --- /dev/null +++ b/cmd/continuity-runtime/internal_executor_test.go @@ -0,0 +1,64 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "git.cin.su/m/Assistant/adapters/postgres" + "git.cin.su/m/Assistant/contracts/continuity" +) + +type fakeContinuityExecutorRepository struct { + receipt continuity.ActionReceipt + err error + calls int +} + +func (f *fakeContinuityExecutorRepository) SatisfyDelivered(context.Context, continuity.SatisfyDeliveredCommand) (continuity.ActionReceipt, error) { + f.calls++ + return f.receipt, f.err +} + +func (f *fakeContinuityExecutorRepository) LoadActionReceipt(context.Context, continuity.CommittedActionIntentID) (continuity.ActionReceipt, error) { + return f.receipt, f.err +} + +func TestInternalExecutorRequiresTokenAndReturnsReceipt(t *testing.T) { + repository := &fakeContinuityExecutorRepository{receipt: continuity.ActionReceipt{ActionIntentID: "intent-1", ResultState: continuity.ProspectiveSatisfied, ResultRevision: 2}} + mux := http.NewServeMux() + registerInternalExecutorRoutes(mux, repository, "secret") + command := continuity.SatisfyDeliveredCommand{ActionIntentID: "intent-1"} + body, _ := json.Marshal(command) + + denied := httptest.NewRequest(http.MethodPost, "/v1/internal/executor/satisfy-delivered", bytes.NewReader(body)) + deniedResult := httptest.NewRecorder() + mux.ServeHTTP(deniedResult, denied) + if deniedResult.Code != http.StatusUnauthorized || repository.calls != 0 { + t.Fatalf("denied status=%d calls=%d", deniedResult.Code, repository.calls) + } + + request := httptest.NewRequest(http.MethodPost, "/v1/internal/executor/satisfy-delivered", bytes.NewReader(body)) + request.Header.Set("Authorization", "Bearer secret") + result := httptest.NewRecorder() + mux.ServeHTTP(result, request) + if result.Code != http.StatusOK || repository.calls != 1 { + t.Fatalf("status=%d calls=%d body=%s", result.Code, repository.calls, result.Body.String()) + } +} + +func TestInternalExecutorMapsOwnerConflictWithoutLeakingDetails(t *testing.T) { + repository := &fakeContinuityExecutorRepository{err: postgres.ErrProspectiveOwnerConflict} + mux := http.NewServeMux() + registerInternalExecutorRoutes(mux, repository, "secret") + request := httptest.NewRequest(http.MethodPost, "/v1/internal/executor/satisfy-delivered", bytes.NewBufferString(`{}`)) + request.Header.Set("Authorization", "Bearer secret") + result := httptest.NewRecorder() + mux.ServeHTTP(result, request) + if result.Code != http.StatusConflict { + t.Fatalf("status=%d", result.Code) + } +} diff --git a/cmd/continuity-runtime/main.go b/cmd/continuity-runtime/main.go index c3877ed..6285341 100644 --- a/cmd/continuity-runtime/main.go +++ b/cmd/continuity-runtime/main.go @@ -30,6 +30,11 @@ func main() { } mux := http.NewServeMux() mux.Handle("/healthz", health.Handler{Runtime: "continuity-runtime", SemanticGeneration: 1, ArchitectureGeneration: 1}) + executorToken := os.Getenv("COMMIT_EXECUTOR_TOKEN") + if executorToken == "" { + log.Fatal("COMMIT_EXECUTOR_TOKEN is required") + } + registerInternalExecutorRoutes(mux, repository, executorToken) mux.HandleFunc("POST /v1/prospective-items", func(w http.ResponseWriter, r *http.Request) { var item continuity.ProspectiveItem if err := json.NewDecoder(r.Body).Decode(&item); err != nil { diff --git a/contracts/commit/types.go b/contracts/commit/types.go index 519b22f..bddf76d 100644 --- a/contracts/commit/types.go +++ b/contracts/commit/types.go @@ -3,6 +3,7 @@ package commit import ( "git.cin.su/m/Assistant/contracts/authority" "git.cin.su/m/Assistant/contracts/capability" + "git.cin.su/m/Assistant/contracts/continuity" "git.cin.su/m/Assistant/contracts/identity" "git.cin.su/m/Assistant/contracts/work" "git.cin.su/m/Assistant/platform/ids" @@ -26,21 +27,32 @@ const ( type ExternalOperation string +const ( + CapabilityTestExecutor capability.CapabilityID = "test-executor" + CapabilityContinuityTransition capability.CapabilityID = "continuity-transition" + OperationTestIdempotentEffect ExternalOperation = "test-idempotent-effect" + OperationSatisfyDelivered ExternalOperation = "satisfy-delivered" +) + type ActionProposal struct { - IntentID ActionIntentID - ActorID identity.ActorID - CapabilityID capability.CapabilityID - Target work.TaskTarget - Operation ExternalOperation - State ActionState + IntentID ActionIntentID + ActorID identity.ActorID + CapabilityID capability.CapabilityID + Target work.TaskTarget + Operation ExternalOperation + ContinuitySubject continuity.SatisfyDeliveredSubject + State ActionState } type AuthorizedAction struct { - ID AuthorizedActionID - Proposal ActionProposal - GrantID authority.GrantID - ScopeID authority.ScopeID - State ActionState + ID AuthorizedActionID + Proposal ActionProposal + PrincipalID identity.PrincipalID + GrantID authority.GrantID + GrantRevision authority.AuthorityRevision + ScopeID authority.ScopeID + AuthorizedAtUnixMillis int64 + State ActionState } type ActionAttempt struct { diff --git a/contracts/continuity/types.go b/contracts/continuity/types.go index 84bd7fe..fc8dbfb 100644 --- a/contracts/continuity/types.go +++ b/contracts/continuity/types.go @@ -10,6 +10,8 @@ import ( type CommitmentID ids.ID type ProspectiveItemID ids.ID type ProspectiveRevision uint64 +type CommittedActionIntentID ids.ID +type CommittedAuthorizedActionID ids.ID type ProspectiveState string @@ -36,3 +38,31 @@ type ProspectiveItem struct { State ProspectiveState Revision ProspectiveRevision } + +// SatisfyDeliveredSubject is the complete serializable operation subject. It +// is distinct from a Work TaskTarget, which remains the live justification. +type SatisfyDeliveredSubject struct { + ProspectiveItemID ProspectiveItemID + ExpectedRevision ProspectiveRevision + FromState ProspectiveState + ToState ProspectiveState + EvidenceRef string +} + +// SatisfyDeliveredCommand is accepted only at Continuity's internal executor +// boundary. Commit identities are references; Continuity remains state owner. +type SatisfyDeliveredCommand struct { + ActionIntentID CommittedActionIntentID + AuthorizedActionID CommittedAuthorizedActionID + OwnerPrincipalID identity.PrincipalID + Subject SatisfyDeliveredSubject +} + +type ActionReceipt struct { + ActionIntentID CommittedActionIntentID + AuthorizedActionID CommittedAuthorizedActionID + OwnerPrincipalID identity.PrincipalID + Subject SatisfyDeliveredSubject + ResultState ProspectiveState + ResultRevision ProspectiveRevision +} diff --git a/database/bootstrap/001_roles_and_schemas.sql b/database/bootstrap/001_roles_and_schemas.sql index afd428c..5a7ad21 100644 --- a/database/bootstrap/001_roles_and_schemas.sql +++ b/database/bootstrap/001_roles_and_schemas.sql @@ -82,7 +82,37 @@ CREATE TABLE IF NOT EXISTS commit.authorized_actions ( CREATE TABLE IF NOT EXISTS commit.action_attempts ( action_attempt_id TEXT PRIMARY KEY, authorized_action_id TEXT NOT NULL, action_intent_id TEXT NOT NULL, state TEXT NOT NULL, - recorded_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP + recorded_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + revision BIGINT NOT NULL DEFAULT 1, + updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + CHECK (revision > 0) +); +CREATE TABLE IF NOT EXISTS commit.authorized_action_envelopes ( + authorized_action_id TEXT PRIMARY KEY, + action_intent_id TEXT NOT NULL UNIQUE, + actor_id TEXT NOT NULL, + principal_id TEXT NOT NULL, + grant_id TEXT NOT NULL, + grant_revision BIGINT NOT NULL, + scope_id TEXT NOT NULL, + task_id TEXT NOT NULL, + goal_id TEXT NOT NULL, + task_scope_id TEXT NOT NULL, + task_revision BIGINT NOT NULL, + capability_id TEXT NOT NULL, + operation TEXT NOT NULL, + prospective_item_id TEXT NOT NULL, + expected_prospective_revision BIGINT NOT NULL, + prospective_from_state TEXT NOT NULL, + prospective_to_state TEXT NOT NULL, + evidence_ref TEXT NOT NULL, + proposal_state TEXT NOT NULL, + action_state TEXT NOT NULL, + authorized_at_unix_millis BIGINT NOT NULL, + envelope_revision BIGINT NOT NULL, + recorded_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + CHECK (grant_revision > 0), CHECK (task_revision > 0), + CHECK (authorized_at_unix_millis > 0), CHECK (envelope_revision = 1) ); RESET ROLE; @@ -142,5 +172,19 @@ CREATE TABLE IF NOT EXISTS continuity.prospective_items ( recorded_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, CHECK (revision > 0) ); +CREATE TABLE IF NOT EXISTS continuity.action_receipts ( + action_intent_id TEXT PRIMARY KEY, + authorized_action_id TEXT NOT NULL, + owner_principal_id TEXT NOT NULL, + prospective_item_id TEXT NOT NULL, + expected_revision BIGINT NOT NULL, + from_state TEXT NOT NULL, + to_state TEXT NOT NULL, + evidence_ref TEXT NOT NULL, + result_state TEXT NOT NULL, + result_revision BIGINT NOT NULL, + recorded_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + CHECK (expected_revision > 0), CHECK (result_revision > expected_revision) +); RESET ROLE; COMMIT; diff --git a/database/migrations/008_commit_authorized_action_envelopes.sql b/database/migrations/008_commit_authorized_action_envelopes.sql new file mode 100644 index 0000000..4466577 --- /dev/null +++ b/database/migrations/008_commit_authorized_action_envelopes.sql @@ -0,0 +1,35 @@ +BEGIN; +SET ROLE commit_role; +ALTER TABLE commit.action_attempts ADD COLUMN IF NOT EXISTS revision BIGINT NOT NULL DEFAULT 1; +ALTER TABLE commit.action_attempts ADD COLUMN IF NOT EXISTS updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP; +CREATE TABLE IF NOT EXISTS commit.authorized_action_envelopes ( + authorized_action_id TEXT PRIMARY KEY, + action_intent_id TEXT NOT NULL UNIQUE, + actor_id TEXT NOT NULL, + principal_id TEXT NOT NULL, + grant_id TEXT NOT NULL, + grant_revision BIGINT NOT NULL, + scope_id TEXT NOT NULL, + task_id TEXT NOT NULL, + goal_id TEXT NOT NULL, + task_scope_id TEXT NOT NULL, + task_revision BIGINT NOT NULL, + capability_id TEXT NOT NULL, + operation TEXT NOT NULL, + prospective_item_id TEXT NOT NULL, + expected_prospective_revision BIGINT NOT NULL, + prospective_from_state TEXT NOT NULL, + prospective_to_state TEXT NOT NULL, + evidence_ref TEXT NOT NULL, + proposal_state TEXT NOT NULL, + action_state TEXT NOT NULL, + authorized_at_unix_millis BIGINT NOT NULL, + envelope_revision BIGINT NOT NULL, + recorded_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + CHECK (grant_revision > 0), + CHECK (task_revision > 0), + CHECK (authorized_at_unix_millis > 0), + CHECK (envelope_revision = 1) +); +RESET ROLE; +COMMIT; diff --git a/database/migrations/009_continuity_action_receipts.sql b/database/migrations/009_continuity_action_receipts.sql new file mode 100644 index 0000000..ffe4905 --- /dev/null +++ b/database/migrations/009_continuity_action_receipts.sql @@ -0,0 +1,19 @@ +BEGIN; +SET ROLE continuity_role; +CREATE TABLE IF NOT EXISTS continuity.action_receipts ( + action_intent_id TEXT PRIMARY KEY, + authorized_action_id TEXT NOT NULL, + owner_principal_id TEXT NOT NULL, + prospective_item_id TEXT NOT NULL, + expected_revision BIGINT NOT NULL, + from_state TEXT NOT NULL, + to_state TEXT NOT NULL, + evidence_ref TEXT NOT NULL, + result_state TEXT NOT NULL, + result_revision BIGINT NOT NULL, + recorded_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + CHECK (expected_revision > 0), + CHECK (result_revision > expected_revision) +); +RESET ROLE; +COMMIT; diff --git a/deploy/docker-compose.dev.yml b/deploy/docker-compose.dev.yml index ad9e367..41fec4a 100644 --- a/deploy/docker-compose.dev.yml +++ b/deploy/docker-compose.dev.yml @@ -65,6 +65,7 @@ services: POSTGRES_USER: continuity_role POSTGRES_PASSWORD: ${CONTINUITY_DB_PASSWORD:-assistant_continuity_dev_only} CORE_RUNTIME_URL: http://core-runtime:8080 + COMMIT_EXECUTOR_TOKEN: ${COMMIT_EXECUTOR_TOKEN:-assistant_executor_dev_only_replace_me} ports: ["19081:8081"] commit-runtime: @@ -82,7 +83,10 @@ services: AUTHORITY_POSTGRES_USER: authority_role AUTHORITY_POSTGRES_PASSWORD: ${AUTHORITY_DB_PASSWORD:-assistant_authority_dev_only} CORE_RUNTIME_URL: http://core-runtime:8080 + CONTINUITY_RUNTIME_URL: http://continuity-runtime:8081 + COMMIT_EXECUTOR_TOKEN: ${COMMIT_EXECUTOR_TOKEN:-assistant_executor_dev_only_replace_me} ports: ["19082:8082"] + depends_on: [core-runtime, continuity-runtime] cognitive-worker: build: diff --git a/deploy/docker-compose.prod.yml b/deploy/docker-compose.prod.yml index eeecd76..0f9eb60 100644 --- a/deploy/docker-compose.prod.yml +++ b/deploy/docker-compose.prod.yml @@ -61,6 +61,7 @@ services: POSTGRES_USER: continuity_role POSTGRES_PASSWORD: ${CONTINUITY_DB_PASSWORD} CORE_RUNTIME_URL: http://core-runtime:8080 + COMMIT_EXECUTOR_TOKEN: ${COMMIT_EXECUTOR_TOKEN} commit-runtime: build: { context: .., dockerfile: deploy/images/runtime.Dockerfile, args: { TARGET: commit-runtime } } @@ -75,7 +76,10 @@ services: AUTHORITY_POSTGRES_USER: authority_role AUTHORITY_POSTGRES_PASSWORD: ${AUTHORITY_DB_PASSWORD} CORE_RUNTIME_URL: http://core-runtime:8080 + CONTINUITY_RUNTIME_URL: http://continuity-runtime:8081 + COMMIT_EXECUTOR_TOKEN: ${COMMIT_EXECUTOR_TOKEN} # External mutation credentials belong here only via Portainer/Docker secrets/env injection. + depends_on: [core-runtime, continuity-runtime] cognitive-worker: build: { context: .., dockerfile: deploy/images/runtime.Dockerfile, args: { TARGET: cognitive-worker } } diff --git a/dev/tasks/DEV-000030-continuity-commit-execution.json b/dev/tasks/DEV-000030-continuity-commit-execution.json new file mode 100644 index 0000000..d7da40f --- /dev/null +++ b/dev/tasks/DEV-000030-continuity-commit-execution.json @@ -0,0 +1,58 @@ +{ + "task_id": "DEV-000030", + "task_revision": 1, + "architecture_revision": "ADR-0003", + "base_repo_revision": "fe0b937", + "risk_tier": "D", + "intent": "Implement the durable typed Commit envelope, the Continuity-owned DELIVERED to SATISFIED command with CAS and idempotency receipt, and remote execution/reconciliation boundaries accepted by ADR-0003.", + "requirement_ids": ["INV-AUTH-001", "INV-OWN-001", "INV-CONT-001", "INV-TOPO-001"], + "allowed_change_scope": { + "modules": ["commit", "continuity"], + "paths": [ + "contracts/commit/**", + "contracts/continuity/**", + "modules/commit/**", + "modules/continuity/**", + "adapters/postgres/commit_repository.go", + "adapters/postgres/continuity_repository.go", + "adapters/continuityexecutor/**", + "cmd/commit-runtime/**", + "cmd/continuity-runtime/**", + "database/bootstrap/001_roles_and_schemas.sql", + "database/migrations/008_commit_authorized_action_envelopes.sql", + "database/migrations/009_continuity_action_receipts.sql", + "deploy/docker-compose.dev.yml", + "deploy/docker-compose.prod.yml", + "tests/action_commit/**", + "tests/prospective_memory/**", + "tests/continuity_execution/**", + "tests/topology_equivalence/**", + "dev/tasks/DEV-000030-continuity-commit-execution.json" + ] + }, + "forbidden_change_scope": { + "paths": ["architecture/**", "security/**", "modules/identity/**", "modules/authority/**", "modules/work/**", "cmd/core-runtime/**", "cmd/creator-web/**"] + }, + "allowed_architecture_delta": {"new_cross_module_dependencies": 0, "new_canonical_owners": 0, "new_privileges": 1}, + "contracts": { + "consume": ["contracts.authority@v1", "contracts.capability@v1", "contracts.work@v1", "contracts.continuity@v1"], + "may_change": ["contracts.commit@v1", "contracts.continuity@v1"] + }, + "semantic_owner": "commit-and-continuity", + "canonical_state": ["Commit authorized envelope and attempts", "Continuity ProspectiveItem and action receipt"], + "state_transitions": ["ProspectiveItem DELIVERED to SATISFIED", "Action AUTHORIZED to EXECUTING to SUCCEEDED, FAILED, or UNKNOWN_OUTCOME"], + "side_effects_and_privileges": ["Commit Runtime calls only the Continuity internal executor API using a runtime-injected shared credential"], + "persistence_and_migrations": ["immutable full Commit envelope", "Continuity receipt keyed by ActionIntentID", "owner-local Prospective CAS"], + "failure_retry_restart": ["transport ambiguity records UNKNOWN_OUTCOME", "no blind retry", "read-only receipt reconciliation", "restart loads exact envelope and receipt"], + "assumptions": ["Creator HTTP orchestration and UI are delivered in later task packets", "existing test executor operation remains supported only for deterministic architecture tests"], + "acceptance_scenarios": [ + "full authorized envelope survives Commit Runtime restart", + "only continuity-transition plus satisfy-delivered can target a Continuity subject", + "Continuity atomically changes an owned DELIVERED item exactly once", + "stale revision, wrong owner, wrong state, or reused mismatched intent cannot mutate state", + "timeout creates UNKNOWN_OUTCOME and reconciliation reads the owner receipt without executing again", + "Commit role cannot write Continuity and Continuity role cannot write Commit" + ], + "required_tests": ["unit", "contract", "repository integration", "runtime HTTP", "adversarial authority", "idempotency", "recovery", "topology equivalence", "DB ACL", "architecture"], + "terminal_gap_states": ["BLOCKED_BY_ARCHITECTURE_GAP", "BLOCKED_BY_SPEC_GAP", "BLOCKED_BY_CONFLICT", "REQUIRES_MIGRATION_DECISION"] +} diff --git a/modules/commit/domain/commit.go b/modules/commit/domain/commit.go index 1403915..7eb4fc7 100644 --- a/modules/commit/domain/commit.go +++ b/modules/commit/domain/commit.go @@ -2,10 +2,12 @@ package domain import ( "errors" + "strings" "git.cin.su/m/Assistant/contracts/authority" "git.cin.su/m/Assistant/contracts/capability" commitcontract "git.cin.su/m/Assistant/contracts/commit" + "git.cin.su/m/Assistant/contracts/continuity" "git.cin.su/m/Assistant/modules/commit/ports" ) @@ -39,14 +41,17 @@ func Reconcile(action commitcontract.AuthorizedAction, previous commitcontract.A } // Authorize is the Commit boundary. A proposal alone never reaches Executor. -func Authorize(proposal commitcontract.ActionProposal, context AuthorizationContext, validator ports.AuthorityValidator, authorizedID commitcontract.AuthorizedActionID) (commitcontract.AuthorizedAction, error) { +func Authorize(proposal commitcontract.ActionProposal, context AuthorizationContext, validator ports.AuthorityValidator, authorizedID commitcontract.AuthorizedActionID, authorizedAtUnixMillis int64) (commitcontract.AuthorizedAction, error) { if proposal.IntentID == "" || proposal.ActorID == "" || proposal.CapabilityID == capability.CapabilityID("") || - proposal.Target.TaskID == "" || proposal.Target.GoalID == "" || proposal.Target.ScopeID == "" || proposal.Target.Revision == 0 || proposal.Operation == "" { + proposal.Target.TaskID == "" || proposal.Target.GoalID == "" || proposal.Target.ScopeID == "" || proposal.Target.Revision == 0 || proposal.Operation == "" || authorizedAtUnixMillis <= 0 { return commitcontract.AuthorizedAction{}, ErrInvalidProposal } if proposal.ActorID != context.Binding.ActorID || proposal.Target.ScopeID != context.Scope.ID { return commitcontract.AuthorizedAction{}, ErrInvalidProposal } + if !allowedOperation(proposal, context) { + return commitcontract.AuthorizedAction{}, ErrInvalidProposal + } if err := validator.Validate(context); err != nil { return commitcontract.AuthorizedAction{}, err } @@ -54,7 +59,21 @@ func Authorize(proposal commitcontract.ActionProposal, context AuthorizationCont return commitcontract.AuthorizedAction{}, ErrInvalidProposal } proposal.State = commitcontract.ActionProposed - return commitcontract.AuthorizedAction{ID: authorizedID, Proposal: proposal, GrantID: context.Grant.ID, ScopeID: context.Scope.ID, State: commitcontract.ActionAuthorized}, nil + return commitcontract.AuthorizedAction{ID: authorizedID, Proposal: proposal, PrincipalID: context.Binding.PrincipalID, GrantID: context.Grant.ID, GrantRevision: context.Grant.Revision, ScopeID: context.Scope.ID, AuthorizedAtUnixMillis: authorizedAtUnixMillis, State: commitcontract.ActionAuthorized}, nil +} + +func allowedOperation(proposal commitcontract.ActionProposal, context AuthorizationContext) bool { + switch { + case proposal.CapabilityID == commitcontract.CapabilityTestExecutor && proposal.Operation == commitcontract.OperationTestIdempotentEffect: + return true + case context.Grant.Class == authority.GrantClassCreator && proposal.CapabilityID == commitcontract.CapabilityContinuityTransition && proposal.Operation == commitcontract.OperationSatisfyDelivered: + subject := proposal.ContinuitySubject + return subject.ProspectiveItemID != "" && subject.ExpectedRevision > 0 && + subject.FromState == continuity.ProspectiveDelivered && subject.ToState == continuity.ProspectiveSatisfied && + strings.TrimSpace(subject.EvidenceRef) != "" + default: + return false + } } // Execute records UNKNOWN_OUTCOME rather than silently retrying an ambiguous diff --git a/modules/continuity/domain/prospective.go b/modules/continuity/domain/prospective.go index 2c52261..b679e77 100644 --- a/modules/continuity/domain/prospective.go +++ b/modules/continuity/domain/prospective.go @@ -2,6 +2,7 @@ package domain import ( "errors" + "strings" contract "git.cin.su/m/Assistant/contracts/continuity" ) @@ -10,12 +11,37 @@ var ( ErrProspectiveNotFound = errors.New("prospective item not found") ErrProspectiveRevision = errors.New("prospective item revision conflict") ErrInvalidTransition = errors.New("invalid prospective item transition") + ErrInvalidSatisfaction = errors.New("invalid delivered satisfaction command") ) type Store struct { items map[contract.ProspectiveItemID]contract.ProspectiveItem } +func ValidateSatisfyDelivered(command contract.SatisfyDeliveredCommand) error { + if command.ActionIntentID == "" || command.AuthorizedActionID == "" || command.OwnerPrincipalID == "" || + command.Subject.ProspectiveItemID == "" || command.Subject.ExpectedRevision == 0 || + command.Subject.FromState != contract.ProspectiveDelivered || command.Subject.ToState != contract.ProspectiveSatisfied || + strings.TrimSpace(command.Subject.EvidenceRef) == "" { + return ErrInvalidSatisfaction + } + return nil +} + +func (s *Store) SatisfyDelivered(id contract.ProspectiveItemID, revision contract.ProspectiveRevision, evidenceRef string) (contract.ProspectiveItem, error) { + item, err := s.current(id, revision) + if err != nil { + return contract.ProspectiveItem{}, err + } + if item.State != contract.ProspectiveDelivered || strings.TrimSpace(evidenceRef) == "" { + return contract.ProspectiveItem{}, ErrInvalidTransition + } + item.State = contract.ProspectiveSatisfied + item.Revision++ + s.items[id] = item + return item, nil +} + func Recover(items []contract.ProspectiveItem) Store { stored := make(map[contract.ProspectiveItemID]contract.ProspectiveItem, len(items)) for _, item := range items { diff --git a/tests/action_commit/continuity_authorization_test.go b/tests/action_commit/continuity_authorization_test.go new file mode 100644 index 0000000..68a8e76 --- /dev/null +++ b/tests/action_commit/continuity_authorization_test.go @@ -0,0 +1,41 @@ +package action_commit_test + +import ( + "errors" + "testing" + + commitcontract "git.cin.su/m/Assistant/contracts/commit" + "git.cin.su/m/Assistant/contracts/continuity" + commitdomain "git.cin.su/m/Assistant/modules/commit/domain" + "git.cin.su/m/Assistant/platform/ids" +) + +func TestContinuityOperationRequiresExactAllowlistedEnvelope(t *testing.T) { + context, proposal := testContextAndProposal() + proposal.CapabilityID = commitcontract.CapabilityContinuityTransition + proposal.Operation = commitcontract.OperationSatisfyDelivered + proposal.ContinuitySubject = continuity.SatisfyDeliveredSubject{ + ProspectiveItemID: "prospective-1", + ExpectedRevision: 4, + FromState: continuity.ProspectiveDelivered, + ToState: continuity.ProspectiveSatisfied, + EvidenceRef: "artifact://evidence/1", + } + action, err := commitdomain.Authorize(proposal, context, validator{}, commitcontract.AuthorizedActionID(ids.ID("authorized-continuity")), 10) + if err != nil { + t.Fatal(err) + } + if action.PrincipalID != context.Binding.PrincipalID || action.GrantRevision != context.Grant.Revision || action.AuthorizedAtUnixMillis != 10 { + t.Fatalf("authorized envelope lost authority evidence: %#v", action) + } + + proposal.Operation = "arbitrary-operation" + if _, err := commitdomain.Authorize(proposal, context, validator{}, "rejected", 10); !errors.Is(err, commitdomain.ErrInvalidProposal) { + t.Fatalf("arbitrary operation error = %v", err) + } + proposal.Operation = commitcontract.OperationSatisfyDelivered + proposal.ContinuitySubject.EvidenceRef = " " + if _, err := commitdomain.Authorize(proposal, context, validator{}, "rejected-evidence", 10); !errors.Is(err, commitdomain.ErrInvalidProposal) { + t.Fatalf("empty evidence error = %v", err) + } +} diff --git a/tests/action_commit/safe_commit_test.go b/tests/action_commit/safe_commit_test.go index a38998d..095e6db 100644 --- a/tests/action_commit/safe_commit_test.go +++ b/tests/action_commit/safe_commit_test.go @@ -5,7 +5,6 @@ import ( "testing" "git.cin.su/m/Assistant/contracts/authority" - "git.cin.su/m/Assistant/contracts/capability" commitcontract "git.cin.su/m/Assistant/contracts/commit" "git.cin.su/m/Assistant/contracts/identity" "git.cin.su/m/Assistant/contracts/work" @@ -30,7 +29,7 @@ func (unknownExecutor) Execute(commitcontract.AuthorizedAction) ports.ExecutionR func TestWS003RejectsRevokedGrantAndNeverExecutesProposalDirectly(t *testing.T) { context, proposal := testContextAndProposal() context.Grant.Active = false - _, err := commitdomain.Authorize(proposal, context, validator{}, commitcontract.AuthorizedActionID(ids.ID("authorized"))) + _, err := commitdomain.Authorize(proposal, context, validator{}, commitcontract.AuthorizedActionID(ids.ID("authorized")), 1) if !errors.Is(err, authoritydomain.ErrInvalidGrant) { t.Fatalf("authorization error = %v, want invalid grant", err) } @@ -39,7 +38,7 @@ func TestWS003RejectsRevokedGrantAndNeverExecutesProposalDirectly(t *testing.T) func TestWS003RejectsMissingActorPrincipalBinding(t *testing.T) { context, proposal := testContextAndProposal() context.Binding.Active = false - _, err := commitdomain.Authorize(proposal, context, validator{}, commitcontract.AuthorizedActionID(ids.ID("authorized"))) + _, err := commitdomain.Authorize(proposal, context, validator{}, commitcontract.AuthorizedActionID(ids.ID("authorized")), 1) if !errors.Is(err, authoritydomain.ErrInvalidGrant) { t.Fatalf("authorization error = %v, want invalid grant", err) } @@ -47,7 +46,7 @@ func TestWS003RejectsMissingActorPrincipalBinding(t *testing.T) { func TestWS003UnknownOutcomeBlocksBlindRetry(t *testing.T) { context, proposal := testContextAndProposal() - action, err := commitdomain.Authorize(proposal, context, validator{}, commitcontract.AuthorizedActionID(ids.ID("authorized"))) + action, err := commitdomain.Authorize(proposal, context, validator{}, commitcontract.AuthorizedActionID(ids.ID("authorized")), 1) if err != nil { t.Fatal(err) } @@ -67,6 +66,6 @@ func testContextAndProposal() (commitdomain.AuthorizationContext, commitcontract scope := authority.Scope{ID: authority.ScopeID(ids.ID("global")), Global: true, Label: "all"} grant := authority.AuthorityGrant{ID: authority.GrantID(ids.ID("grant")), ActorID: actor, PrincipalID: principal, ScopeID: scope.ID, Class: authority.GrantClassCreator, Revision: 1, Active: true} context := commitdomain.AuthorizationContext{Binding: identity.PrincipalBinding{ActorID: actor, PrincipalID: principal, Active: true}, Scope: scope, Grant: grant} - proposal := commitcontract.ActionProposal{IntentID: commitcontract.ActionIntentID(ids.ID("intent")), ActorID: actor, CapabilityID: capability.CapabilityID(ids.ID("executor")), Target: work.TaskTarget{TaskID: work.TaskID(ids.ID("task")), GoalID: work.GoalID(ids.ID("goal")), ScopeID: scope.ID, Revision: 1}, Operation: "mutate-external-world"} + proposal := commitcontract.ActionProposal{IntentID: commitcontract.ActionIntentID(ids.ID("intent")), ActorID: actor, CapabilityID: commitcontract.CapabilityTestExecutor, Target: work.TaskTarget{TaskID: work.TaskID(ids.ID("task")), GoalID: work.GoalID(ids.ID("goal")), ScopeID: scope.ID, Revision: 1}, Operation: commitcontract.OperationTestIdempotentEffect} return context, proposal } diff --git a/tests/action_commit/test_executor_reconciliation_test.go b/tests/action_commit/test_executor_reconciliation_test.go index f4e99c0..09637a3 100644 --- a/tests/action_commit/test_executor_reconciliation_test.go +++ b/tests/action_commit/test_executor_reconciliation_test.go @@ -12,7 +12,7 @@ import ( func TestWS003UnknownOutcomeReconcilesWithoutDuplicateEffect(t *testing.T) { context, proposal := testContextAndProposal() - action, err := commitdomain.Authorize(proposal, context, validator{}, commitcontract.AuthorizedActionID(ids.ID("authorized-reconcile"))) + action, err := commitdomain.Authorize(proposal, context, validator{}, commitcontract.AuthorizedActionID(ids.ID("authorized-reconcile")), 1) if err != nil { t.Fatal(err) } @@ -34,7 +34,7 @@ func TestWS003UnknownOutcomeReconcilesWithoutDuplicateEffect(t *testing.T) { func TestWS003ReconciliationRequiresUnknownOutcome(t *testing.T) { context, proposal := testContextAndProposal() - action, err := commitdomain.Authorize(proposal, context, validator{}, commitcontract.AuthorizedActionID(ids.ID("authorized-invalid-reconcile"))) + action, err := commitdomain.Authorize(proposal, context, validator{}, commitcontract.AuthorizedActionID(ids.ID("authorized-invalid-reconcile")), 1) if err != nil { t.Fatal(err) } diff --git a/tests/continuity_execution/verify-remote.ps1 b/tests/continuity_execution/verify-remote.ps1 new file mode 100644 index 0000000..00c8429 --- /dev/null +++ b/tests/continuity_execution/verify-remote.ps1 @@ -0,0 +1,115 @@ +param([string]$DockerContext = "docker-prod") + +$ErrorActionPreference = "Stop" + +function Invoke-JsonPost([string]$container, [string]$url, [object]$body, [bool]$internal = $false) { + $json = $body | ConvertTo-Json -Depth 10 -Compress + $local = New-TemporaryFile + $remote = "/tmp/domovoy-$([guid]::NewGuid().ToString('N')).json" + try { + [IO.File]::WriteAllText($local, $json, [Text.UTF8Encoding]::new($false)) + docker --context $DockerContext cp $local "${container}:$remote" | Out-Null + if ($internal) { + $response = docker --context $DockerContext exec $container sh -c 'wget -qO- --header="Content-Type: application/json" --header="Authorization: Bearer $COMMIT_EXECUTOR_TOKEN" --post-file "$1" "$2"' sh $remote $url + } else { + $response = docker --context $DockerContext exec $container wget -qO- --header "Content-Type: application/json" --post-file $remote $url + } + if ($LASTEXITCODE -ne 0) { throw "POST $url failed" } + return ($response -join "`n") + } finally { + docker --context $DockerContext exec --user root $container rm -f $remote 2>$null + Remove-Item -LiteralPath $local -Force + } +} + +function Expect-InternalPostStatus([string]$container, [string]$url, [object]$body, [int]$status) { + $json = $body | ConvertTo-Json -Depth 10 -Compress + $local = New-TemporaryFile + $remote = "/tmp/domovoy-$([guid]::NewGuid().ToString('N')).json" + try { + [IO.File]::WriteAllText($local, $json, [Text.UTF8Encoding]::new($false)) + docker --context $DockerContext cp $local "${container}:$remote" | Out-Null + $previous = $ErrorActionPreference + $ErrorActionPreference = "Continue" + $response = (docker --context $DockerContext exec $container sh -c 'wget -S -O /dev/null --header="Content-Type: application/json" --header="Authorization: Bearer $COMMIT_EXECUTOR_TOKEN" --post-file "$1" "$2"' sh $remote $url 2>&1) -join "`n" + $ErrorActionPreference = $previous + if ($response -notmatch "HTTP/1\.[01] $status") { throw "POST $url did not return $status" } + } finally { + docker --context $DockerContext exec --user root $container rm -f $remote 2>$null + Remove-Item -LiteralPath $local -Force + } +} + +function Internal-Get([string]$container, [string]$url) { + $response = docker --context $DockerContext exec $container sh -c 'wget -qO- --header="Authorization: Bearer $COMMIT_EXECUTOR_TOKEN" "$1"' sh $url + if ($LASTEXITCODE -ne 0) { throw "GET $url failed" } + return ($response -join "`n") +} + +$suffix = [guid]::NewGuid().ToString("N") +$scope = "domovoy-global-scope" +$goal = "goal-execution-$suffix" +$task = "task-execution-$suffix" +$prospective = "prospective-execution-$suffix" +$intent = "intent-execution-$suffix" +$authorized = "authorized-execution-$suffix" +$attempt = "attempt-execution-$suffix" +$evidence = "artifact://execution/$suffix" + +Invoke-JsonPost assistant-core-runtime-1 http://localhost:8080/v1/tasks @{ + TaskID=$task; GoalID=$goal; ScopeID=$scope; Revision=1 +} | Out-Null +Invoke-JsonPost assistant-continuity-runtime-1 http://localhost:8081/v1/prospective-items @{ + ID=$prospective; CommitmentID="commitment-$suffix"; OwnerPrincipalID="domovoy-creator-principal" + ScopeID=$scope; GoalID=$goal; TriggerAtUnixMillis=1800000000000; State="DELIVERED"; Revision=1 +} | Out-Null +Invoke-JsonPost assistant-commit-runtime-1 http://localhost:8082/v1/authorized-actions @{ + Proposal=@{ + IntentID=$intent; ActorID="domovoy-creator-actor"; CapabilityID="continuity-transition" + Target=@{TaskID=$task;GoalID=$goal;ScopeID=$scope;Revision=1}; Operation="satisfy-delivered" + ContinuitySubject=@{ProspectiveItemID=$prospective;ExpectedRevision=1;FromState="DELIVERED";ToState="SATISFIED";EvidenceRef=$evidence} + } + AuthorizedID=$authorized;GrantID="domovoy-creator-grant";PrincipalID="domovoy-creator-principal" +} | Out-Null + +# Prove the executable envelope is sufficient after process loss. +docker --context $DockerContext restart assistant-commit-runtime-1 | Out-Null +for ($index=0; $index -lt 15; $index++) { + docker --context $DockerContext exec assistant-commit-runtime-1 wget -qO- http://localhost:8082/healthz 2>$null | Out-Null + if ($LASTEXITCODE -eq 0) { break } + Start-Sleep -Seconds 1 +} +if ($LASTEXITCODE -ne 0) { throw "commit runtime did not recover" } + +$attemptResult = Invoke-JsonPost assistant-commit-runtime-1 "http://localhost:8082/v1/internal/authorized-actions/$authorized/execute" @{AttemptID=$attempt} $true | ConvertFrom-Json +if ($attemptResult.State -ne "SUCCEEDED") { throw "execution did not succeed" } +$item = docker --context $DockerContext exec assistant-continuity-runtime-1 wget -qO- "http://localhost:8081/v1/prospective-items/$prospective" | ConvertFrom-Json +if ($item.State -ne "SATISFIED" -or $item.Revision -ne 2) { throw "ProspectiveItem did not transition exactly once" } + +# Exact redelivery is answered from the owner receipt and cannot increment again. +$sameCommand = @{ + ActionIntentID=$intent;AuthorizedActionID=$authorized;OwnerPrincipalID="domovoy-creator-principal" + Subject=@{ProspectiveItemID=$prospective;ExpectedRevision=1;FromState="DELIVERED";ToState="SATISFIED";EvidenceRef=$evidence} +} +$receipt = Invoke-JsonPost assistant-continuity-runtime-1 http://localhost:8081/v1/internal/executor/satisfy-delivered $sameCommand $true | ConvertFrom-Json +if ($receipt.ResultRevision -ne 2) { throw "idempotent receipt revision changed" } +$itemAfterReplay = docker --context $DockerContext exec assistant-continuity-runtime-1 wget -qO- "http://localhost:8081/v1/prospective-items/$prospective" | ConvertFrom-Json +if ($itemAfterReplay.Revision -ne 2) { throw "duplicate delivery changed canonical revision" } +Expect-InternalPostStatus assistant-commit-runtime-1 "http://localhost:8082/v1/internal/authorized-actions/$authorized/execute" @{AttemptID="duplicate-$suffix"} 409 + +# A new intent with a stale revision cannot mutate the already satisfied item. +$stale = @{ + ActionIntentID="stale-$suffix";AuthorizedActionID="stale-authorized-$suffix";OwnerPrincipalID="domovoy-creator-principal" + Subject=@{ProspectiveItemID=$prospective;ExpectedRevision=1;FromState="DELIVERED";ToState="SATISFIED";EvidenceRef="artifact://stale/$suffix"} +} +Expect-InternalPostStatus assistant-continuity-runtime-1 http://localhost:8081/v1/internal/executor/satisfy-delivered $stale 409 + +docker --context $DockerContext restart assistant-continuity-runtime-1 | Out-Null +for ($index=0; $index -lt 15; $index++) { + $receiptAfterRestart = Internal-Get assistant-continuity-runtime-1 "http://localhost:8081/v1/internal/executor/receipts/$intent" 2>$null + if ($LASTEXITCODE -eq 0) { break } + Start-Sleep -Seconds 1 +} +if (($receiptAfterRestart | ConvertFrom-Json).ResultRevision -ne 2) { throw "receipt did not survive restart" } + +Write-Output "Continuity Commit execution verification: PASS" diff --git a/tests/prospective_memory/satisfaction_test.go b/tests/prospective_memory/satisfaction_test.go new file mode 100644 index 0000000..a3550c2 --- /dev/null +++ b/tests/prospective_memory/satisfaction_test.go @@ -0,0 +1,32 @@ +package prospective_memory_test + +import ( + "errors" + "testing" + + "git.cin.su/m/Assistant/contracts/continuity" + continuitydomain "git.cin.su/m/Assistant/modules/continuity/domain" +) + +func TestDeliveredItemCanBeSatisfiedExactlyAtExpectedRevision(t *testing.T) { + item := continuity.ProspectiveItem{ID: "prospective-1", State: continuity.ProspectiveDelivered, Revision: 4} + store := continuitydomain.Recover([]continuity.ProspectiveItem{item}) + updated, err := store.SatisfyDelivered(item.ID, item.Revision, "artifact://evidence/1") + if err != nil || updated.State != continuity.ProspectiveSatisfied || updated.Revision != 5 { + t.Fatalf("satisfaction = %#v, %v", updated, err) + } + if _, err := store.SatisfyDelivered(item.ID, item.Revision, "artifact://evidence/1"); !errors.Is(err, continuitydomain.ErrProspectiveRevision) { + t.Fatalf("stale satisfaction error = %v", err) + } +} + +func TestSatisfactionRejectsMissingEvidenceAndWrongState(t *testing.T) { + store := continuitydomain.Recover([]continuity.ProspectiveItem{{ID: "prospective-1", State: continuity.ProspectiveArmed, Revision: 1}}) + if _, err := store.SatisfyDelivered("prospective-1", 1, "evidence"); !errors.Is(err, continuitydomain.ErrInvalidTransition) { + t.Fatalf("wrong state error = %v", err) + } + store = continuitydomain.Recover([]continuity.ProspectiveItem{{ID: "prospective-2", State: continuity.ProspectiveDelivered, Revision: 1}}) + if _, err := store.SatisfyDelivered("prospective-2", 1, " "); !errors.Is(err, continuitydomain.ErrInvalidTransition) { + t.Fatalf("missing evidence error = %v", err) + } +}