feat(store): add TaskCorrected compensating-event type (S8)
Implements §3.1's invariant that a wrong event is never edited, only compensated for by a new appended event. TaskCorrected references the event it repairs and can change state and/or amend-style fields; Store.Append verifies the referenced event actually exists on the task. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01W1rkJ2hBMybnJctPbcy4tT
This commit is contained in:
@@ -101,7 +101,7 @@ func ValidateEvent(e Event) error {
|
||||
if e.SchemaVersion >= 2 && strings.TrimSpace(e.Surface) == "" {
|
||||
return fmt.Errorf("%w: surface required", 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}
|
||||
allowed := map[string]bool{"TaskCreated": true, "TaskLeased": true, "TaskReleased": true, "TaskCompleted": true, "TaskFailed": true, "TaskBlocked": true, "ApprovalRequested": true, "ApprovalGranted": true, "ApprovalDenied": true, "TaskAmended": true, "TaskCorrected": true, "QuotaReported": true, "StandupAdvisory": true}
|
||||
if !allowed[e.Type] {
|
||||
return fmt.Errorf("%w: unknown type %q", ErrInvalid, e.Type)
|
||||
}
|
||||
@@ -189,6 +189,29 @@ func ValidatePayload(typ string, p map[string]any) error {
|
||||
if len(p) == 0 {
|
||||
return fmt.Errorf("%w: amendment cannot be empty", ErrInvalid)
|
||||
}
|
||||
case "TaskCorrected":
|
||||
// §3.1: "a wrong event is never edited; a compensating event is
|
||||
// appended and replay sees both." `corrects` names the event this one
|
||||
// reverses/repairs — existence against the log is checked in
|
||||
// Store.Append, where the log is visible; ValidatePayload only knows
|
||||
// shape.
|
||||
if err := requiredString("corrects"); err != nil {
|
||||
return err
|
||||
}
|
||||
if v, ok := p["state"]; ok {
|
||||
s, ok := v.(string)
|
||||
if !ok {
|
||||
return fmt.Errorf("%w: state must be a string", ErrInvalid)
|
||||
}
|
||||
switch TaskState(s) {
|
||||
case StateQueued, StateLeased, StateCompleted, StateFailed, StateBlocked:
|
||||
default:
|
||||
return fmt.Errorf("%w: state invalid", ErrInvalid)
|
||||
}
|
||||
}
|
||||
if len(p) < 2 {
|
||||
return fmt.Errorf("%w: correction must change at least one field", ErrInvalid)
|
||||
}
|
||||
case "ApprovalRequested":
|
||||
for _, k := range []string{"subject_ref", "options"} {
|
||||
if _, ok := p[k]; !ok {
|
||||
|
||||
@@ -169,6 +169,27 @@ func (s *Store) apply(e domain.Event) error {
|
||||
t.Due = &d
|
||||
}
|
||||
}
|
||||
case "TaskCorrected":
|
||||
if v, ok := p["title"].(string); ok {
|
||||
t.Title = v
|
||||
}
|
||||
if v, ok := p["description"].(string); ok {
|
||||
t.Description = v
|
||||
}
|
||||
if v, ok := p["inherent_priority"].(float64); ok {
|
||||
t.InherentPriority = int(v)
|
||||
}
|
||||
if v, ok := p["due"].(string); ok {
|
||||
if d, err := time.Parse(time.RFC3339, v); err == nil {
|
||||
t.Due = &d
|
||||
}
|
||||
}
|
||||
if v, ok := p["state"].(string); ok {
|
||||
t.State = domain.TaskState(v)
|
||||
if t.State != domain.StateLeased {
|
||||
t.Lease = nil
|
||||
}
|
||||
}
|
||||
}
|
||||
t.Version = e.Version
|
||||
s.tasks[e.TaskID] = t
|
||||
@@ -230,6 +251,21 @@ func (s *Store) Append(e domain.Event) error {
|
||||
return domain.ErrConflict
|
||||
}
|
||||
}
|
||||
if e.Type == "TaskCorrected" {
|
||||
var p map[string]any
|
||||
_ = json.Unmarshal(e.Payload, &p)
|
||||
corrects, _ := p["corrects"].(string)
|
||||
found := false
|
||||
for _, prior := range s.events {
|
||||
if prior.ID == corrects && prior.TaskID == e.TaskID {
|
||||
found = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
return fmt.Errorf("%w: corrects references unknown event %q for this task", domain.ErrInvalid, corrects)
|
||||
}
|
||||
}
|
||||
global := e.Type == "QuotaReported" || e.Type == "StandupAdvisory" || e.Type == "ApprovalGranted" || e.Type == "ApprovalDenied"
|
||||
if !taskExists && e.Type != "TaskCreated" && !global {
|
||||
return domain.ErrNotFound
|
||||
|
||||
@@ -86,6 +86,56 @@ func TestTaskAmendedAppliesAllFields(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestTaskCorrected guards S8: the spec (§3.1) requires corrections to be
|
||||
// compensating events appended on top of a wrong one, never an edit of the
|
||||
// log — a wrongly-emitted terminal state (e.g. a mistaken TaskFailed) must be
|
||||
// repairable by appending a new event that references the one it corrects,
|
||||
// with both surviving in the replayed log.
|
||||
func TestTaskCorrected(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
s, err := Open(dir)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := s.Append(created("e1")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
failPayload, _ := json.Marshal(map[string]any{"reason": "mistaken failure"})
|
||||
failEvt := domain.Event{ID: "e2", Type: "TaskFailed", TaskID: "task-1", Version: 2, Payload: failPayload, Surface: string(authz.System)}
|
||||
if err := s.Append(failEvt); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if tk := s.Tasks()[0]; tk.State != domain.StateFailed {
|
||||
t.Fatalf("expected failed, got %s", tk.State)
|
||||
}
|
||||
|
||||
// Referencing an unknown event is rejected.
|
||||
badCorrection, _ := json.Marshal(map[string]any{"corrects": "does-not-exist", "state": "queued"})
|
||||
if err := s.Append(domain.Event{Type: "TaskCorrected", TaskID: "task-1", Version: 3, Payload: badCorrection, Surface: string(authz.System)}); !errors.Is(err, domain.ErrInvalid) {
|
||||
t.Fatalf("expected ErrInvalid for unknown corrects target, got %v", err)
|
||||
}
|
||||
|
||||
correction, _ := json.Marshal(map[string]any{"corrects": "e2", "state": "queued"})
|
||||
if err := s.Append(domain.Event{ID: "e3", Type: "TaskCorrected", TaskID: "task-1", Version: 3, Payload: correction, Surface: string(authz.System)}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
tk := s.Tasks()[0]
|
||||
if tk.State != domain.StateQueued {
|
||||
t.Fatalf("expected correction to restore queued state, got %s", tk.State)
|
||||
}
|
||||
|
||||
// Replay from disk still applies both the wrong event and its correction
|
||||
// (the snapshot only elides already-applied events from Events(), never
|
||||
// from the durable log itself).
|
||||
s2, err := Open(dir)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if tk2 := s2.Tasks()[0]; tk2.State != domain.StateQueued {
|
||||
t.Fatalf("expected replayed state queued, got %s", tk2.State)
|
||||
}
|
||||
}
|
||||
|
||||
// TestLeaseAndExpireEventIDsAreUnique guards S5: Event.ID was set to the
|
||||
// task id in both Lease and ExpireLeases, so every lease of the same task
|
||||
// produced a TaskLeased/TaskReleased event with a colliding ID — unsound
|
||||
|
||||
Reference in New Issue
Block a user