This commit is contained in:
@@ -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}
|
||||
}
|
||||
@@ -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"},
|
||||
},
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
+23
-11
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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;
|
||||
@@ -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;
|
||||
@@ -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:
|
||||
|
||||
@@ -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 } }
|
||||
|
||||
@@ -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"]
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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"
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user