diff --git a/internal/domain/domain.go b/internal/domain/domain.go index 7cc13ca..c6fccc3 100644 --- a/internal/domain/domain.go +++ b/internal/domain/domain.go @@ -17,6 +17,8 @@ var ErrConflict = errors.New("task version conflict") var ErrNotFound = errors.New("task not found") var ErrInvalid = errors.New("invalid event") +const CurrentEventSchema = 1 + type TaskState string const ( @@ -53,13 +55,14 @@ type Task struct { } type Event struct { - Seq uint64 `json:"seq"` - ID string `json:"id"` - Type string `json:"type"` - TaskID string `json:"task_id"` - Version int `json:"version"` - At time.Time `json:"at"` - Payload json.RawMessage `json:"payload"` + SchemaVersion int `json:"schema_version,omitempty"` + Seq uint64 `json:"seq"` + ID string `json:"id"` + Type string `json:"type"` + TaskID string `json:"task_id"` + Version int `json:"version"` + At time.Time `json:"at"` + Payload json.RawMessage `json:"payload"` } func Hash(v []byte) string { h := sha256.Sum256(v); return hex.EncodeToString(h[:]) } @@ -76,7 +79,7 @@ func NewID() string { return ulidEncoding.EncodeToString(b) } func ValidateEvent(e Event) error { - if e.Type == "" || e.TaskID == "" || len(e.Payload) == 0 || len(e.Payload) > 64*1024 { + if e.SchemaVersion > CurrentEventSchema || e.Type == "" || e.TaskID == "" || len(e.Payload) == 0 || len(e.Payload) > 64*1024 { return ErrInvalid } allowed := map[string]bool{"TaskCreated": true, "TaskLeased": true, "TaskReleased": true, "TaskCompleted": true, "TaskFailed": true, "TaskBlocked": true, "ApprovalRequested": true, "ApprovalGranted": true, "ApprovalDenied": true, "TaskAmended": true, "QuotaReported": true, "StandupAdvisory": true} @@ -87,6 +90,9 @@ func ValidateEvent(e Event) error { if err := json.Unmarshal(e.Payload, &p); err != nil { return fmt.Errorf("%w: payload is not JSON", ErrInvalid) } + if p == nil { + return fmt.Errorf("%w: payload must be an object", ErrInvalid) + } return ValidatePayload(e.Type, p) } func ValidateCreated(p map[string]any) error { @@ -116,6 +122,9 @@ func ValidatePayload(typ string, p map[string]any) error { if _, ok := p["until_ns"].(float64); !ok { return fmt.Errorf("%w: until_ns required", ErrInvalid) } + if v, ok := p["expected_version"].(float64); ok && v < 0 { + return fmt.Errorf("%w: expected_version invalid", ErrInvalid) + } case "TaskReleased": if err := requiredString("handoff_ref"); err != nil && p["reason"] == nil { return err diff --git a/internal/store/store.go b/internal/store/store.go index 1d8d665..7a727c4 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -166,6 +166,9 @@ func (s *Store) Append(e domain.Event) error { if e.Seq == 0 { e.Seq = s.seq + 1 } + if e.SchemaVersion == 0 { + e.SchemaVersion = domain.CurrentEventSchema + } if e.Type == "TaskCreated" { var p map[string]any if err := json.Unmarshal(e.Payload, &p); err != nil { @@ -175,10 +178,22 @@ func (s *Store) Append(e domain.Event) error { return nil } } - if t, ok := s.tasks[e.TaskID]; ok && e.Version != t.Version+1 { + t, taskExists := s.tasks[e.TaskID] + if taskExists && e.Version != t.Version+1 { return domain.ErrConflict } - if _, ok := s.tasks[e.TaskID]; !ok && e.Type != "TaskCreated" { + if e.Type == "TaskLeased" { + var p struct { + ExpectedVersion *int `json:"expected_version"` + } + if err := json.Unmarshal(e.Payload, &p); err != nil { + return err + } + if p.ExpectedVersion != nil && (t.Version != *p.ExpectedVersion) { + return domain.ErrConflict + } + } + if !taskExists && e.Type != "TaskCreated" { return domain.ErrNotFound } if e.Type != "TaskCreated" && (e.Type == "TaskCompleted" || e.Type == "TaskBlocked" || e.Type == "TaskReleased") { @@ -293,7 +308,7 @@ func (s *Store) Lease(id, harness string, ttl time.Duration) (domain.Event, erro if t.State != domain.StateQueued { return domain.Event{}, domain.ErrConflict } - p, _ := json.Marshal(map[string]any{"harness_id": harness, "until_ns": time.Now().Add(ttl).UnixNano()}) + p, _ := json.Marshal(map[string]any{"harness_id": harness, "until_ns": time.Now().Add(ttl).UnixNano(), "expected_version": t.Version}) e := domain.Event{ID: id, Type: "TaskLeased", TaskID: id, Version: t.Version + 1, Payload: p} return e, s.Append(e) } diff --git a/progress.md b/progress.md index f3b7391..f521a7e 100644 --- a/progress.md +++ b/progress.md @@ -53,6 +53,8 @@ The latest pass now applies `AuthorizeEvent` to lifecycle and approval writes an The lifecycle API contract pass now requires callers to provide explicit release evidence (`reason` or `handoff_ref`), a block `blocker`, and a completion `report_ref`. The server no longer creates generated placeholder completion artifacts or converts malformed/empty lifecycle bodies into defaults. Regression coverage validates that release, completion, and block events reject missing evidence. +Item 1 substrate hardening pass: newly appended events use schema envelope version 1; replay rejects unsupported versions and non-object payloads; and `TaskLeased` carries an `expected_version` guard that is checked before append. Legacy envelope events remain readable for tolerant replay. + Recommended order: 1. Add the orchestration coordinator: lease → worktree → harness session → bootstrap → lifecycle events.