From 64c7082e49f0c53a508da9ed8e8193dd353c90f2 Mon Sep 17 00:00:00 2001 From: Janitha Karunarathna Date: Sat, 10 Oct 2026 23:04:24 +0530 Subject: [PATCH 1/8] feat(taskflow)!: add task claims to the task store interface TaskRecord gains ClaimedBy and ClaimedAt, and TaskStore gains ClaimTask and ReleaseTask: guarded writes that set or clear the claim without touching seq or the step columns. The holder is an opaque value chosen by the host; taskflow does not decide who may claim. BREAKING CHANGE: TaskStore implementations must add ClaimTask and ReleaseTask. --- taskflow/store/db.go | 24 ++++++++++++++++++++++++ taskflow/store/db_test.go | 24 ++++++++++++++++++++++++ 2 files changed, 48 insertions(+) diff --git a/taskflow/store/db.go b/taskflow/store/db.go index d0214f2..2af2b2b 100644 --- a/taskflow/store/db.go +++ b/taskflow/store/db.go @@ -49,6 +49,14 @@ type TaskRecord struct { // Data holds generic, dynamic task execution state variables. Data map[string]any `json:"data"` + // ClaimedBy is who holds the task's claim, or "" when it is unclaimed. It is an opaque value + // chosen by the host (for example an internal user ID); taskflow never interprets it or decides + // who may claim. It is written only by ClaimTask and ReleaseTask, never by InitTask or the step + // writes, and it is kept after the task completes. + ClaimedBy string `json:"claimed_by,omitempty"` + // ClaimedAt is when ClaimedBy first claimed the task; nil when unclaimed. + ClaimedAt *time.Time `json:"claimed_at,omitempty"` + CreatedAt time.Time `json:"created_at"` UpdatedAt time.Time `json:"updated_at"` } @@ -61,6 +69,10 @@ func (r TaskRecord) DeepCopy() TaskRecord { cp := r // value copy duplicates all scalar fields cp.RenderConfig = copyBytes(r.RenderConfig) cp.Data = deepcopy.Map(r.Data) + if r.ClaimedAt != nil { + at := *r.ClaimedAt + cp.ClaimedAt = &at + } return cp } @@ -104,6 +116,11 @@ type StepClaim struct { // part of the write, so a write from a step that is no longer current changes nothing. Each returns // the number of rows changed: 0 means the write was stale and was dropped, which is not an error. // InitTask must not write ActiveStepID or Seq. +// +// ClaimTask and ReleaseTask are conditional writes of the same kind on the claim columns +// (ClaimedBy, ClaimedAt). They are independent of the step columns: they never change Seq, so a +// claim does not make an open step stale, and the step writes never change the claim. InitTask +// must not write the claim columns either. type TaskStore interface { // InitTask creates the row for a new task (state, parent coordinates, render config snapshot). // It is called once, by StartTask, but must be idempotent: StartTask is a Temporal Activity and @@ -129,4 +146,11 @@ type TaskStore interface { PersistSubmission(context context.Context, taskID, stepID string, seq int64, data map[string]any) (int64, error) // CompleteTask sets StateCompleted and seq. Guard: the stored seq is <= seq. CompleteTask(context context.Context, taskID string, seq int64) (int64, error) + + // ClaimTask sets the claim to holder. Guard: the task is unclaimed or already claimed by + // holder, and its state is not StateCompleted. A repeat claim by the holder keeps the + // original ClaimedAt; otherwise ClaimedAt is set to at. + ClaimTask(context context.Context, taskID, holder string, at time.Time) (int64, error) + // ReleaseTask clears the claim. Guard: the task is claimed by holder. + ReleaseTask(context context.Context, taskID, holder string) (int64, error) } diff --git a/taskflow/store/db_test.go b/taskflow/store/db_test.go index 40b8ab9..5a35146 100644 --- a/taskflow/store/db_test.go +++ b/taskflow/store/db_test.go @@ -56,6 +56,30 @@ func (t *testStore) PersistSubmission(context.Context, string, string, int64, ma func (t *testStore) CompleteTask(context.Context, string, int64) (int64, error) { return 0, nil } +func (t *testStore) ClaimTask(context.Context, string, string, time.Time) (int64, error) { + return 0, nil +} + +func (t *testStore) ReleaseTask(context.Context, string, string) (int64, error) { return 0, nil } + +// The copy must not share ClaimedAt with the original, as it shares no other reference state. +func TestDeepCopy_CopiesClaimedAt(t *testing.T) { + at := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) + original := TaskRecord{TaskID: "test-1", ClaimedBy: "user-1", ClaimedAt: &at} + + cp := original.DeepCopy() + if cp.ClaimedBy != "user-1" || cp.ClaimedAt == nil || !cp.ClaimedAt.Equal(at) { + t.Fatalf("claim not copied: got (%q, %v)", cp.ClaimedBy, cp.ClaimedAt) + } + if cp.ClaimedAt == original.ClaimedAt { + t.Fatal("ClaimedAt is shared with the original") + } + + if got := (TaskRecord{TaskID: "test-2"}).DeepCopy().ClaimedAt; got != nil { + t.Errorf("unclaimed copy ClaimedAt = %v, want nil", got) + } +} + func TestTaskStoreInterface(t *testing.T) { var store TaskStore = &testStore{tasks: make(map[string]TaskRecord)} From 6379483af85c8d639fb4db28ec71c9d5394efd7f Mon Sep 17 00:00:00 2001 From: Janitha Karunarathna Date: Sat, 10 Oct 2026 23:14:29 +0530 Subject: [PATCH 2/8] feat(taskflow): store task claims in the gorm task store Add claimed_by and claimed_at to TaskRecordModel and implement ClaimTask and ReleaseTask as single guarded UPDATEs. A claim succeeds when the task is unclaimed or already held by the caller, and is refused once the task is completed; a repeat claim keeps the original claimed_at. InitTask omits the claim columns so a StartTask retry cannot reset a claim. Neither write touches seq or the step columns. Hosts must add the columns: ALTER TABLE task_records_v2 ADD COLUMN claimed_by TEXT NULL; ALTER TABLE task_records_v2 ADD COLUMN claimed_at TIMESTAMPTZ NULL; --- taskflow/store/gorm/model.go | 22 ++++- taskflow/store/gorm/model_test.go | 18 ++++ taskflow/store/gorm/store.go | 27 ++++- taskflow/store/gorm/store_test.go | 157 ++++++++++++++++++++++++++++++ 4 files changed, 219 insertions(+), 5 deletions(-) diff --git a/taskflow/store/gorm/model.go b/taskflow/store/gorm/model.go index 483d39b..f3cda71 100644 --- a/taskflow/store/gorm/model.go +++ b/taskflow/store/gorm/model.go @@ -28,8 +28,12 @@ type TaskRecordModel struct { ActiveStepID *string `gorm:"column:active_step_id;type:uuid"` Seq int64 `gorm:"column:seq;not null;default:0"` Data json.RawMessage `gorm:"column:data;type:jsonb;serializer:json"` - CreatedAt time.Time `gorm:"column:created_at;type:timestamptz;not null;autoCreateTime"` - UpdatedAt time.Time `gorm:"column:updated_at;type:timestamptz;not null;autoUpdateTime"` + // ClaimedBy and ClaimedAt belong to ClaimTask and ReleaseTask in store.go. InitTask never + // writes them. The table needs: claimed_by TEXT NULL, claimed_at TIMESTAMPTZ NULL. + ClaimedBy *string `gorm:"column:claimed_by;type:text"` + ClaimedAt *time.Time `gorm:"column:claimed_at;type:timestamptz"` + CreatedAt time.Time `gorm:"column:created_at;type:timestamptz;not null;autoCreateTime"` + UpdatedAt time.Time `gorm:"column:updated_at;type:timestamptz;not null;autoUpdateTime"` } func (TaskRecordModel) TableName() string { @@ -51,6 +55,11 @@ func (m TaskRecordModel) ToDomain() store.TaskRecord { activeStepID = *m.ActiveStepID } + var claimedBy string + if m.ClaimedBy != nil { + claimedBy = *m.ClaimedBy + } + return store.TaskRecord{ TaskID: m.TaskID, TaskType: m.TaskType, @@ -64,6 +73,8 @@ func (m TaskRecordModel) ToDomain() store.TaskRecord { ActiveStepID: activeStepID, Seq: m.Seq, Data: data, + ClaimedBy: claimedBy, + ClaimedAt: m.ClaimedAt, CreatedAt: m.CreatedAt, UpdatedAt: m.UpdatedAt, } @@ -81,6 +92,11 @@ func FromDomain(r store.TaskRecord) TaskRecordModel { activeStepID = &r.ActiveStepID } + var claimedBy *string + if r.ClaimedBy != "" { + claimedBy = &r.ClaimedBy + } + return TaskRecordModel{ TaskID: r.TaskID, TaskType: r.TaskType, @@ -94,5 +110,7 @@ func FromDomain(r store.TaskRecord) TaskRecordModel { ActiveStepID: activeStepID, Seq: r.Seq, Data: dataBytes, + ClaimedBy: claimedBy, + ClaimedAt: r.ClaimedAt, } } diff --git a/taskflow/store/gorm/model_test.go b/taskflow/store/gorm/model_test.go index 703cdc7..3846cd5 100644 --- a/taskflow/store/gorm/model_test.go +++ b/taskflow/store/gorm/model_test.go @@ -7,6 +7,7 @@ import ( "encoding/json" "reflect" "testing" + "time" "github.com/OpenNSW/core/taskflow/store" ) @@ -21,6 +22,7 @@ import ( // CreatedAt/UpdatedAt are intentionally excluded: GORM owns them via // autoCreateTime/autoUpdateTime, so FromDomain never copies them. func TestRoundTrip_PreservesFields(t *testing.T) { + claimedAt := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) original := store.TaskRecord{ TaskID: "task-1", TaskType: "TEST", @@ -34,6 +36,8 @@ func TestRoundTrip_PreservesFields(t *testing.T) { ActiveStepID: "0a0a0a0a-0000-4000-8000-00000000000a", Seq: 4, Data: map[string]any{"userform": map[string]any{"name": "Alice"}}, + ClaimedBy: "user-1", + ClaimedAt: &claimedAt, } got := FromDomain(original).ToDomain() @@ -62,6 +66,20 @@ func TestRoundTrip_PreservesFields(t *testing.T) { if !reflect.DeepEqual(got.Data, original.Data) { t.Errorf("Data not preserved: got %#v, want %#v", got.Data, original.Data) } + if got.ClaimedBy != original.ClaimedBy || got.ClaimedAt == nil || !got.ClaimedAt.Equal(claimedAt) { + t.Errorf("claim columns: got (%q, %v), want (%q, %v)", got.ClaimedBy, got.ClaimedAt, original.ClaimedBy, claimedAt) + } +} + +// An unclaimed task must reach the claim columns as NULL, not as "", so "IS NULL" matches it. +func TestFromDomain_UnclaimedIsNull(t *testing.T) { + m := FromDomain(store.TaskRecord{TaskID: "task-1"}) + if m.ClaimedBy != nil || m.ClaimedAt != nil { + t.Errorf("claim columns = (%v, %v), want (nil, nil)", m.ClaimedBy, m.ClaimedAt) + } + if got := m.ToDomain(); got.ClaimedBy != "" || got.ClaimedAt != nil { + t.Errorf("round trip claim = (%q, %v), want empty", got.ClaimedBy, got.ClaimedAt) + } } // TestFromDomain_DoesNotDeriveCaseID locks in that FromDomain is a pure diff --git a/taskflow/store/gorm/store.go b/taskflow/store/gorm/store.go index 5b96f31..7736dd5 100644 --- a/taskflow/store/gorm/store.go +++ b/taskflow/store/gorm/store.go @@ -9,6 +9,7 @@ import ( "errors" "fmt" "log/slog" + "time" "github.com/OpenNSW/core/taskflow/store" "gorm.io/gorm" @@ -21,20 +22,23 @@ type TaskStore struct { db *gorm.DB } +var _ store.TaskStore = (*TaskStore)(nil) + func New(db *gorm.DB) *TaskStore { return &TaskStore{db: db} } func (s *TaskStore) InitTask(ctx context.Context, record store.TaskRecord) error { - // active_step_id and seq are owned by the guarded step statements below. Omitting them keeps a - // full-record save from moving them, and lets this write work before the columns exist. + // active_step_id and seq are owned by the guarded step statements below, and claimed_by and + // claimed_at by ClaimTask and ReleaseTask. Omitting them keeps a full-record save from moving + // them, and lets this write work before the columns exist. model := FromDomain(record) // On conflict, do nothing rather than overwrite: TaskID is the parent's ActivationID, unique // per invocation of the parent's node, so a conflict can only be a retry of this exact // StartTask call (Temporal Activities can be retried) — never a different task. The existing // row is already correct, and by the time a retry lands, guarded writes below (ClaimStep etc.) // may have already moved it forward; overwriting state/data here would silently rewind it. - err := s.db.WithContext(ctx).Omit("active_step_id", "seq").Clauses(clause.OnConflict{ + err := s.db.WithContext(ctx).Omit("active_step_id", "seq", "claimed_by", "claimed_at").Clauses(clause.OnConflict{ Columns: []clause.Column{{Name: "task_id"}}, DoNothing: true, }).Create(&model).Error @@ -152,3 +156,20 @@ func (s *TaskStore) CompleteTask(ctx context.Context, taskID string, seq int64) "seq": seq, }) } + +func (s *TaskStore) ClaimTask(ctx context.Context, taskID, holder string, at time.Time) (int64, error) { + // The holder matches the guard again on a repeat claim, which then succeeds without moving + // claimed_at: the CASE reads the row's value before this update, as SQL assignments do. + return s.stepUpdate(ctx, taskID, "(claimed_by IS NULL OR claimed_by = ?) AND state <> ?", + []any{holder, store.StateCompleted}, map[string]any{ + "claimed_by": holder, + "claimed_at": gorm.Expr("CASE WHEN claimed_by = ? THEN claimed_at ELSE ? END", holder, at), + }) +} + +func (s *TaskStore) ReleaseTask(ctx context.Context, taskID, holder string) (int64, error) { + return s.stepUpdate(ctx, taskID, "claimed_by = ?", []any{holder}, map[string]any{ + "claimed_by": nil, + "claimed_at": nil, + }) +} diff --git a/taskflow/store/gorm/store_test.go b/taskflow/store/gorm/store_test.go index f20cdd7..a35e107 100644 --- a/taskflow/store/gorm/store_test.go +++ b/taskflow/store/gorm/store_test.go @@ -5,8 +5,12 @@ package gorm import ( "context" + "fmt" "reflect" + "sync" + "sync/atomic" "testing" + "time" "github.com/glebarez/sqlite" "gorm.io/gorm" @@ -44,6 +48,7 @@ func newTestStore(t *testing.T) (*TaskStore, context.Context) { task_workflow_id TEXT, active_task_template_id TEXT, case_id TEXT NOT NULL DEFAULT '', active_step_id TEXT NULL, seq INTEGER NOT NULL DEFAULT 0, data TEXT, + claimed_by TEXT NULL, claimed_at TIMESTAMP NULL, created_at TIMESTAMP NOT NULL, updated_at TIMESTAMP NOT NULL)`).Error; err != nil { t.Fatalf("create table: %v", err) } @@ -202,6 +207,158 @@ func TestGuardedWrites_UnknownTaskChangesNothing(t *testing.T) { wantRows(t, n, err, 0) n, err = s.CompleteTask(ctx, "nope", 1) wantRows(t, n, err, 0) + n, err = s.ClaimTask(ctx, "nope", userA, t1) + wantRows(t, n, err, 0) + n, err = s.ReleaseTask(ctx, "nope", userA) + wantRows(t, n, err, 0) +} + +var ( + t1 = time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) + t2 = t1.Add(time.Hour) +) + +const ( + userA = "user-a" + userB = "user-b" +) + +func wantClaim(t *testing.T, got store.TaskRecord, holder string, at *time.Time) { + t.Helper() + if got.ClaimedBy != holder { + t.Fatalf("ClaimedBy = %q, want %q", got.ClaimedBy, holder) + } + switch { + case at == nil && got.ClaimedAt != nil: + t.Fatalf("ClaimedAt = %v, want nil", got.ClaimedAt) + case at != nil && (got.ClaimedAt == nil || !got.ClaimedAt.Equal(*at)): + t.Fatalf("ClaimedAt = %v, want %v", got.ClaimedAt, *at) + } +} + +func TestClaimTask(t *testing.T) { + s, ctx := newTestStore(t) + wantClaim(t, mustGet(t, s, ctx), "", nil) // a new task is unclaimed + + n, err := s.ClaimTask(ctx, task1, userA, t1) + wantRows(t, n, err, 1) + wantClaim(t, mustGet(t, s, ctx), userA, &t1) + + // A repeat claim by the holder succeeds and keeps the original claimed_at. + n, err = s.ClaimTask(ctx, task1, userA, t2) + wantRows(t, n, err, 1) + wantClaim(t, mustGet(t, s, ctx), userA, &t1) + + // Another user cannot take a held claim. + n, err = s.ClaimTask(ctx, task1, userB, t2) + wantRows(t, n, err, 0) + wantClaim(t, mustGet(t, s, ctx), userA, &t1) +} + +func TestReleaseTask(t *testing.T) { + s, ctx := newTestStore(t) + _, _ = s.ClaimTask(ctx, task1, userA, t1) + + // Only the holder can release. + n, err := s.ReleaseTask(ctx, task1, userB) + wantRows(t, n, err, 0) + wantClaim(t, mustGet(t, s, ctx), userA, &t1) + + n, err = s.ReleaseTask(ctx, task1, userA) + wantRows(t, n, err, 1) + wantClaim(t, mustGet(t, s, ctx), "", nil) + + // Releasing an unclaimed task changes nothing; the store leaves it to the caller to decide + // whether that is an error. + n, err = s.ReleaseTask(ctx, task1, userA) + wantRows(t, n, err, 0) + + // Once released, anyone can claim, and claimed_at is the new claim's time. + n, err = s.ClaimTask(ctx, task1, userB, t2) + wantRows(t, n, err, 1) + wantClaim(t, mustGet(t, s, ctx), userB, &t2) +} + +// A completed task cannot be claimed, but a claim held when it completed is kept as a record of +// who worked it, and the holder can still release it. +func TestClaimTask_CompletedTask(t *testing.T) { + s, ctx := newTestStore(t) + _, _ = s.ClaimTask(ctx, task1, userA, t1) + _, _ = s.CompleteTask(ctx, task1, 1) + wantClaim(t, mustGet(t, s, ctx), userA, &t1) + + n, err := s.ClaimTask(ctx, task1, userA, t2) + wantRows(t, n, err, 0) + n, err = s.ClaimTask(ctx, task1, userB, t2) + wantRows(t, n, err, 0) + wantClaim(t, mustGet(t, s, ctx), userA, &t1) + + n, err = s.ReleaseTask(ctx, task1, userA) + wantRows(t, n, err, 1) + n, err = s.ClaimTask(ctx, task1, userB, t2) + wantRows(t, n, err, 0) // still completed +} + +// Claims and step writes own separate columns: a claim must not make an open step stale, and the +// step writes must not drop the claim. +func TestClaimAndStepWrites_DoNotTouchEachOther(t *testing.T) { + s, ctx := newTestStore(t) + _, _ = s.ClaimStep(ctx, task1, claim(stepA, 4, nil)) + _, _ = s.WriteRenderState(ctx, task1, stepA, 4, "PENDING_USER", map[string]any{"form": "v1"}) + + n, err := s.ClaimTask(ctx, task1, userA, t1) + wantRows(t, n, err, 1) + got := mustGet(t, s, ctx) + wantRow(t, got, stepA, 4, "PENDING_USER") + if want := map[string]any{"form": "v1"}; !reflect.DeepEqual(got.Data, want) { + t.Errorf("Data = %v, want %v", got.Data, want) + } + + // The step is submitted, the next step starts and the task completes, all under the claim. + _, _ = s.PersistSubmission(ctx, task1, stepA, 4, map[string]any{"a": "submitted"}) + _, _ = s.ClaimStep(ctx, task1, claim(stepB, 5, nil)) + _, _ = s.WriteRenderState(ctx, task1, stepB, 5, "PENDING_USER", nil) + _, _ = s.CompleteTask(ctx, task1, 6) + wantRow(t, mustGet(t, s, ctx), stepB, 6, store.StateCompleted) + wantClaim(t, mustGet(t, s, ctx), userA, &t1) + + // InitTask (a StartTask retry) does not reset the claim either. + s.InitTask(ctx, store.TaskRecord{TaskID: task1, TaskType: "TEST", State: "OLD", Data: map[string]any{}}) + wantClaim(t, mustGet(t, s, ctx), userA, &t1) +} + +// Many users claim at once and exactly one wins. The test database has one connection, so the +// claims are serialised; this checks that the guard admits a single holder, while Postgres's +// row lock does the same for truly concurrent updates. +func TestClaimTask_ConcurrentClaimsHaveOneWinner(t *testing.T) { + s, ctx := newTestStore(t) + + const n = 20 + var ( + wg sync.WaitGroup + wins atomic.Int32 + ) + for i := range n { + wg.Add(1) + go func() { + defer wg.Done() + rows, err := s.ClaimTask(ctx, task1, fmt.Sprintf("user-%d", i), t1) + if err != nil { + t.Errorf("claim %d: %v", i, err) + } + if rows == 1 { + wins.Add(1) + } + }() + } + wg.Wait() + + if got := wins.Load(); got != 1 { + t.Fatalf("%d claims won, want exactly 1", got) + } + if mustGet(t, s, ctx).ClaimedBy == "" { + t.Fatal("task is unclaimed after the race") + } } // The end-to-end example from the design doc: A runs at seq 4 and B is the final step. It From 3d6ec3cf9e309b4c665bfb44834dc4a686bceecd Mon Sep 17 00:00:00 2001 From: Janitha Karunarathna Date: Sat, 10 Oct 2026 23:21:19 +0530 Subject: [PATCH 3/8] feat(taskflow): claim and release tasks through the TaskManager Add TaskManager.ClaimTask and ReleaseTask. When a guarded write matches no row, the manager reads the row and returns a typed error: ErrTaskNotFound, ErrTaskCompleted, a *ClaimHeldError naming the current holder (matches ErrClaimHeld), or ErrNotClaimHolder. Releasing an unclaimed task succeeds. If the claim changes between the write and that read, the call is retried a bounded number of times. Taskflow only stores claims; hosts decide who may claim and what a claim is required for. CompleteTaskStep and GetTaskRenderInfo now wrap the new ErrTaskNotFound, so hosts can answer a missing task with 404 instead of 500. --- taskflow/orchestrator/claim.go | 122 ++++++++++++ taskflow/orchestrator/claim_test.go | 262 ++++++++++++++++++++++++++ taskflow/orchestrator/manager.go | 7 +- taskflow/orchestrator/manager_test.go | 27 +++ 4 files changed, 416 insertions(+), 2 deletions(-) create mode 100644 taskflow/orchestrator/claim.go create mode 100644 taskflow/orchestrator/claim_test.go diff --git a/taskflow/orchestrator/claim.go b/taskflow/orchestrator/claim.go new file mode 100644 index 0000000..3f803af --- /dev/null +++ b/taskflow/orchestrator/claim.go @@ -0,0 +1,122 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright (c) 2026 Lanka Software Foundation + +package orchestrator + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/OpenNSW/core/taskflow/store" +) + +// A claim records who is working a task, so two people do not act on it at once. The holder is an +// opaque value chosen by the host (for example an internal user ID). taskflow only stores claims: +// it never decides who may claim a task or what a claim is required for. That is the host's +// policy, enforced in its own handlers before it calls CompleteTaskStep. +// +// Not to be confused with store.TaskStore.ClaimStep, which makes a step the task's active step, or +// with renderer.Facts.Claims, which are authorization facts for rendering. + +// ErrHolderRequired is returned by ClaimTask and ReleaseTask when the holder is empty. A caller +// serving HTTP should answer it with 400; it usually means the caller has no identity to claim as. +var ErrHolderRequired = errors.New("claim holder is required") + +// ErrClaimHeld means another holder has the task's claim. ClaimTask returns it as a +// *ClaimHeldError, which names the holder. A caller serving HTTP should answer it with 409. +var ErrClaimHeld = errors.New("task is claimed by another holder") + +// ErrNotClaimHolder is returned by ReleaseTask when the task is claimed by someone else. A caller +// serving HTTP should answer it with 403. +var ErrNotClaimHolder = errors.New("task is not claimed by the caller") + +// ErrTaskCompleted is returned by ClaimTask when the task has completed: there is nothing left to +// work on. A caller serving HTTP should answer it with 409. +var ErrTaskCompleted = errors.New("task is completed") + +// ClaimHeldError is the error ClaimTask returns when another holder has the claim. It matches +// ErrClaimHeld with errors.Is. +type ClaimHeldError struct { + // Holder is who has the claim. + Holder string + // ClaimedAt is when Holder claimed the task, if known. + ClaimedAt *time.Time +} + +func (e *ClaimHeldError) Error() string { + return fmt.Sprintf("%s: %s", ErrClaimHeld, e.Holder) +} + +func (e *ClaimHeldError) Is(target error) bool { return target == ErrClaimHeld } + +// claimAttempts bounds the retries of a claim or release whose guard failed because the claim +// changed between the write and the read that explains it (for example, the holder released in +// between). Each retry needs another concurrent change to fail again, so a few are plenty. +const claimAttempts = 3 + +// ClaimTask claims the task for holder. It succeeds when the task is unclaimed or already claimed +// by holder; a repeat claim keeps the original claim time. The task's steps and version are not +// changed, so an open step stays current. +// +// Errors: ErrHolderRequired, ErrTaskNotFound, ErrTaskCompleted (the task has completed, even if +// holder had claimed it), or a *ClaimHeldError (another holder has the claim). +func (tm *TaskManager) ClaimTask(ctx context.Context, taskID, holder string) error { + if holder == "" { + return ErrHolderRequired + } + for range claimAttempts { + n, err := tm.db.ClaimTask(ctx, taskID, holder, time.Now().UTC()) + if err != nil { + return fmt.Errorf("claiming task %s: %w", taskID, err) + } + if n > 0 { + return nil + } + + // The guard did not match: read the row to say why. + record, ok := tm.db.GetTask(ctx, taskID) + switch { + case !ok: + return fmt.Errorf("%w: %s", ErrTaskNotFound, taskID) + case record.State == store.StateCompleted: + return fmt.Errorf("%w: %s", ErrTaskCompleted, taskID) + case record.ClaimedBy != "" && record.ClaimedBy != holder: + return &ClaimHeldError{Holder: record.ClaimedBy, ClaimedAt: record.ClaimedAt} + } + // The row is claimable now, so the claim changed after the write: try again. + } + return fmt.Errorf("claiming task %s: the claim kept changing, try again", taskID) +} + +// ReleaseTask clears holder's claim on the task. Releasing a task that is not claimed succeeds, so +// a repeated release is harmless. A completed task's claim can still be released. +// +// Errors: ErrHolderRequired, ErrTaskNotFound, or ErrNotClaimHolder (someone else has the claim). +func (tm *TaskManager) ReleaseTask(ctx context.Context, taskID, holder string) error { + if holder == "" { + return ErrHolderRequired + } + for range claimAttempts { + n, err := tm.db.ReleaseTask(ctx, taskID, holder) + if err != nil { + return fmt.Errorf("releasing task %s: %w", taskID, err) + } + if n > 0 { + return nil + } + + record, ok := tm.db.GetTask(ctx, taskID) + switch { + case !ok: + return fmt.Errorf("%w: %s", ErrTaskNotFound, taskID) + case record.ClaimedBy == "": + return nil + case record.ClaimedBy != holder: + return fmt.Errorf("%w: %s", ErrNotClaimHolder, taskID) + } + // holder has the claim now, so it changed after the write: try again. + } + return fmt.Errorf("releasing task %s: the claim kept changing, try again", taskID) +} diff --git a/taskflow/orchestrator/claim_test.go b/taskflow/orchestrator/claim_test.go new file mode 100644 index 0000000..8001fd6 --- /dev/null +++ b/taskflow/orchestrator/claim_test.go @@ -0,0 +1,262 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright (c) 2026 Lanka Software Foundation + +package orchestrator + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/OpenNSW/core/taskflow/store" +) + +const ( + claimTaskID = "claim-task" + holderA = "user-a" + holderB = "user-b" +) + +// claimFixture is an unclaimed task waiting for the user. +func claimFixture(t *testing.T, db store.TaskStore) *TaskManager { + t.Helper() + if err := db.InitTask(context.Background(), store.TaskRecord{ + TaskID: claimTaskID, TaskType: "TEST", State: "PENDING_USER", Data: map[string]any{}, + }); err != nil { + t.Fatal(err) + } + return newTestTaskManager(db, newTestRegistry(), &mockTemporalManager{}, noopCallback) +} + +func claimRow(t *testing.T, db store.TaskStore) store.TaskRecord { + t.Helper() + row, ok := db.GetTask(context.Background(), claimTaskID) + if !ok { + t.Fatal("task row missing") + } + return row +} + +func TestClaimTask_ClaimsAnUnclaimedTask(t *testing.T) { + db := newSafeMockTaskStore() + tm := claimFixture(t, db) + before := time.Now().UTC() + + if err := tm.ClaimTask(context.Background(), claimTaskID, holderA); err != nil { + t.Fatalf("ClaimTask: %v", err) + } + + row := claimRow(t, db) + if row.ClaimedBy != holderA { + t.Errorf("ClaimedBy = %q, want %q", row.ClaimedBy, holderA) + } + if row.ClaimedAt == nil || row.ClaimedAt.Before(before) { + t.Errorf("ClaimedAt = %v, want a time at or after %v", row.ClaimedAt, before) + } +} + +func TestClaimTask_RepeatClaimByTheHolderSucceeds(t *testing.T) { + db := newSafeMockTaskStore() + tm := claimFixture(t, db) + ctx := context.Background() + if err := tm.ClaimTask(ctx, claimTaskID, holderA); err != nil { + t.Fatal(err) + } + first := *claimRow(t, db).ClaimedAt + + if err := tm.ClaimTask(ctx, claimTaskID, holderA); err != nil { + t.Fatalf("repeat ClaimTask: %v", err) + } + if got := claimRow(t, db).ClaimedAt; got == nil || !got.Equal(first) { + t.Errorf("ClaimedAt = %v, want the original %v", got, first) + } +} + +func TestClaimTask_HeldByAnotherHolder(t *testing.T) { + db := newSafeMockTaskStore() + tm := claimFixture(t, db) + ctx := context.Background() + if err := tm.ClaimTask(ctx, claimTaskID, holderA); err != nil { + t.Fatal(err) + } + + err := tm.ClaimTask(ctx, claimTaskID, holderB) + + if !errors.Is(err, ErrClaimHeld) { + t.Fatalf("err = %v, want ErrClaimHeld", err) + } + var held *ClaimHeldError + if !errors.As(err, &held) || held.Holder != holderA || held.ClaimedAt == nil { + t.Fatalf("err = %#v, want a *ClaimHeldError naming %q with its claim time", err, holderA) + } + if got := claimRow(t, db).ClaimedBy; got != holderA { + t.Errorf("ClaimedBy = %q, want %q unchanged", got, holderA) + } +} + +// A completed task cannot be claimed, even by the holder of a claim kept from before it +// completed. +func TestClaimTask_CompletedTask(t *testing.T) { + db := newSafeMockTaskStore() + tm := claimFixture(t, db) + ctx := context.Background() + if err := tm.ClaimTask(ctx, claimTaskID, holderA); err != nil { + t.Fatal(err) + } + if _, err := db.CompleteTask(ctx, claimTaskID, 1); err != nil { + t.Fatal(err) + } + + for _, holder := range []string{holderA, holderB} { + if err := tm.ClaimTask(ctx, claimTaskID, holder); !errors.Is(err, ErrTaskCompleted) { + t.Errorf("ClaimTask(%q) err = %v, want ErrTaskCompleted", holder, err) + } + } + if got := claimRow(t, db).ClaimedBy; got != holderA { + t.Errorf("ClaimedBy = %q, want the kept claim %q", got, holderA) + } +} + +func TestReleaseTask(t *testing.T) { + db := newSafeMockTaskStore() + tm := claimFixture(t, db) + ctx := context.Background() + if err := tm.ClaimTask(ctx, claimTaskID, holderA); err != nil { + t.Fatal(err) + } + + if err := tm.ReleaseTask(ctx, claimTaskID, holderB); !errors.Is(err, ErrNotClaimHolder) { + t.Fatalf("release by another holder: err = %v, want ErrNotClaimHolder", err) + } + if got := claimRow(t, db).ClaimedBy; got != holderA { + t.Fatalf("ClaimedBy = %q, want %q unchanged", got, holderA) + } + + if err := tm.ReleaseTask(ctx, claimTaskID, holderA); err != nil { + t.Fatalf("release by the holder: %v", err) + } + if row := claimRow(t, db); row.ClaimedBy != "" || row.ClaimedAt != nil { + t.Fatalf("claim = (%q, %v), want cleared", row.ClaimedBy, row.ClaimedAt) + } + + // Releasing an unclaimed task is harmless, so a repeated release succeeds. + if err := tm.ReleaseTask(ctx, claimTaskID, holderA); err != nil { + t.Errorf("repeat release: %v", err) + } + if err := tm.ReleaseTask(ctx, claimTaskID, holderB); err != nil { + t.Errorf("release of an unclaimed task: %v", err) + } +} + +func TestReleaseTask_CompletedTask(t *testing.T) { + db := newSafeMockTaskStore() + tm := claimFixture(t, db) + ctx := context.Background() + if err := tm.ClaimTask(ctx, claimTaskID, holderA); err != nil { + t.Fatal(err) + } + if _, err := db.CompleteTask(ctx, claimTaskID, 1); err != nil { + t.Fatal(err) + } + + if err := tm.ReleaseTask(ctx, claimTaskID, holderA); err != nil { + t.Fatalf("ReleaseTask: %v", err) + } + if got := claimRow(t, db).ClaimedBy; got != "" { + t.Errorf("ClaimedBy = %q, want cleared", got) + } +} + +func TestClaimAndRelease_RequireAHolder(t *testing.T) { + db := newSafeMockTaskStore() + tm := claimFixture(t, db) + ctx := context.Background() + + if err := tm.ClaimTask(ctx, claimTaskID, ""); !errors.Is(err, ErrHolderRequired) { + t.Errorf("ClaimTask err = %v, want ErrHolderRequired", err) + } + if err := tm.ReleaseTask(ctx, claimTaskID, ""); !errors.Is(err, ErrHolderRequired) { + t.Errorf("ReleaseTask err = %v, want ErrHolderRequired", err) + } + if got := claimRow(t, db).ClaimedBy; got != "" { + t.Errorf("ClaimedBy = %q, want unclaimed", got) + } +} + +func TestClaimAndRelease_UnknownTask(t *testing.T) { + tm := claimFixture(t, newSafeMockTaskStore()) + ctx := context.Background() + + if err := tm.ClaimTask(ctx, "nope", holderA); !errors.Is(err, ErrTaskNotFound) { + t.Errorf("ClaimTask err = %v, want ErrTaskNotFound", err) + } + if err := tm.ReleaseTask(ctx, "nope", holderA); !errors.Is(err, ErrTaskNotFound) { + t.Errorf("ReleaseTask err = %v, want ErrTaskNotFound", err) + } +} + +// staleClaimStore drops the first `misses` claim and release writes as if their guard had not +// matched, while the row is in fact claimable: what the manager sees when the claim changes +// between its write and the read that explains a miss. +type staleClaimStore struct { + *safeMockTaskStore + misses int +} + +func (s *staleClaimStore) ClaimTask(ctx context.Context, taskID, holder string, at time.Time) (int64, error) { + if s.misses > 0 { + s.misses-- + return 0, nil + } + return s.safeMockTaskStore.ClaimTask(ctx, taskID, holder, at) +} + +func (s *staleClaimStore) ReleaseTask(ctx context.Context, taskID, holder string) (int64, error) { + if s.misses > 0 { + s.misses-- + return 0, nil + } + return s.safeMockTaskStore.ReleaseTask(ctx, taskID, holder) +} + +func TestClaimAndRelease_RetryWhenTheClaimChangesUnderThem(t *testing.T) { + ctx := context.Background() + + db := &staleClaimStore{safeMockTaskStore: newSafeMockTaskStore()} + tm := claimFixture(t, db) + db.misses = claimAttempts - 1 + if err := tm.ClaimTask(ctx, claimTaskID, holderA); err != nil { + t.Fatalf("ClaimTask after %d misses: %v", claimAttempts-1, err) + } + db.misses = claimAttempts - 1 + if err := tm.ReleaseTask(ctx, claimTaskID, holderA); err != nil { + t.Fatalf("ReleaseTask after %d misses: %v", claimAttempts-1, err) + } + + // With holderA holding the claim, a miss is unexplained for both: holderA may release it and + // may claim it again. After claimAttempts misses they give up with an error. + if err := tm.ClaimTask(ctx, claimTaskID, holderA); err != nil { + t.Fatal(err) + } + db.misses = claimAttempts + if err := tm.ReleaseTask(ctx, claimTaskID, holderA); err == nil { + t.Error("ReleaseTask: want an error after every attempt missed") + } + db.misses = claimAttempts + if err := tm.ClaimTask(ctx, claimTaskID, holderA); err == nil { + t.Error("ClaimTask: want an error after every attempt missed") + } +} + +func TestErrTaskNotFound_FromTaskReads(t *testing.T) { + tm := claimFixture(t, newSafeMockTaskStore()) + ctx := context.Background() + + if _, err := tm.GetTaskRenderInfo(ctx, "nope"); !errors.Is(err, ErrTaskNotFound) { + t.Errorf("GetTaskRenderInfo err = %v, want ErrTaskNotFound", err) + } + if err := tm.CompleteTaskStep(ctx, "nope", stepA, map[string]any{}); !errors.Is(err, ErrTaskNotFound) { + t.Errorf("CompleteTaskStep err = %v, want ErrTaskNotFound", err) + } +} diff --git a/taskflow/orchestrator/manager.go b/taskflow/orchestrator/manager.go index 7c5dd1a..5ea6b26 100644 --- a/taskflow/orchestrator/manager.go +++ b/taskflow/orchestrator/manager.go @@ -87,6 +87,9 @@ var ErrStaleStep = errors.New("step is no longer the active step of the task") // completing. A caller serving HTTP should answer it with 400. var ErrStepIDRequired = errors.New("step ID is required") +// ErrTaskNotFound means no task has the given ID. A caller serving HTTP should answer it with 404. +var ErrTaskNotFound = errors.New("task not found") + // TaskManager orchestrates decoupled tasks and interactions under parent workflows. // It bridges macro-level workflows and micro-level interactive tasks via a single DB entry per task. type TaskManager struct { @@ -390,7 +393,7 @@ func (tm *TaskManager) CompleteTaskStep(ctx context.Context, taskID, stepID stri record, exists := tm.db.GetTask(ctx, taskID) if !exists { - return fmt.Errorf("task %s not found", taskID) + return fmt.Errorf("%w: %s", ErrTaskNotFound, taskID) } if record.State == store.StateCompleted { return fmt.Errorf("%w: task %s already completed", ErrStaleStep, taskID) @@ -530,7 +533,7 @@ func (tm *TaskManager) runExtensions(ctx context.Context, record *store.TaskReco func (tm *TaskManager) GetTaskRenderInfo(context context.Context, taskID string) (TaskView, error) { record, exists := tm.db.GetTask(context, taskID) if !exists { - return TaskView{}, fmt.Errorf("task record %s not found", taskID) + return TaskView{}, fmt.Errorf("%w: %s", ErrTaskNotFound, taskID) } view, err := tm.renderer.Render(context, record.RenderConfig, renderer.Facts{State: record.State, Data: record.Data}) diff --git a/taskflow/orchestrator/manager_test.go b/taskflow/orchestrator/manager_test.go index 859a953..4697474 100644 --- a/taskflow/orchestrator/manager_test.go +++ b/taskflow/orchestrator/manager_test.go @@ -9,6 +9,7 @@ import ( "errors" "sync" "testing" + "time" "github.com/OpenNSW/core/artifact" "github.com/OpenNSW/core/artifact/testutil" @@ -196,6 +197,32 @@ func (s *safeMockTaskStore) CompleteTask(_ context.Context, taskID string, seq i return 1, nil } +func (s *safeMockTaskStore) ClaimTask(_ context.Context, taskID, holder string, at time.Time) (int64, error) { + s.mu.Lock() + defer s.mu.Unlock() + t, ok := s.tasks[taskID] + if !ok || (t.ClaimedBy != "" && t.ClaimedBy != holder) || t.State == store.StateCompleted { + return 0, nil + } + if t.ClaimedBy != holder { + t.ClaimedBy, t.ClaimedAt = holder, &at + } + s.tasks[taskID] = t + return 1, nil +} + +func (s *safeMockTaskStore) ReleaseTask(_ context.Context, taskID, holder string) (int64, error) { + s.mu.Lock() + defer s.mu.Unlock() + t, ok := s.tasks[taskID] + if !ok || t.ClaimedBy != holder { + return 0, nil + } + t.ClaimedBy, t.ClaimedAt = "", nil + s.tasks[taskID] = t + return 1, nil +} + // --------------------------------------------------------------------------- // Helpers // --------------------------------------------------------------------------- From cad514801b33b5b9a4b497b408d69f9dc67eb6d6 Mon Sep 17 00:00:00 2001 From: Janitha Karunarathna Date: Sat, 10 Oct 2026 23:25:24 +0530 Subject: [PATCH 4/8] feat(taskflow): show the task's claim on task views TaskView and ZoneView gain claimed_by and claimed_at, taken from the task record by GetTaskRenderInfo, GetAllTasks and the zone view assembler. Both are omitted when the task is unclaimed. The value is the holder the host recorded; hosts resolve it to a display name. --- taskflow/orchestrator/manager.go | 4 ++ taskflow/orchestrator/manager_test.go | 41 +++++++++++++++++++ taskflow/orchestrator/view.go | 6 ++- taskflow/renderer/zoneview/zone_assembler.go | 2 + .../renderer/zoneview/zone_assembler_test.go | 30 ++++++++++++++ taskflow/renderer/zoneview/zoneview.go | 9 +++- 6 files changed, 89 insertions(+), 3 deletions(-) diff --git a/taskflow/orchestrator/manager.go b/taskflow/orchestrator/manager.go index 5ea6b26..838faaf 100644 --- a/taskflow/orchestrator/manager.go +++ b/taskflow/orchestrator/manager.go @@ -547,6 +547,8 @@ func (tm *TaskManager) GetTaskRenderInfo(context context.Context, taskID string) State: record.State, StepID: record.ActiveStepID, Version: record.Seq, + ClaimedBy: record.ClaimedBy, + ClaimedAt: record.ClaimedAt, CreatedAt: record.CreatedAt, UpdatedAt: record.UpdatedAt, View: view, // actually attach the render output @@ -571,6 +573,8 @@ func (tm *TaskManager) GetAllTasks(ctx context.Context, caseID string) []TaskVie State: r.State, StepID: r.ActiveStepID, Version: r.Seq, + ClaimedBy: r.ClaimedBy, + ClaimedAt: r.ClaimedAt, CreatedAt: r.CreatedAt, UpdatedAt: r.UpdatedAt, }) diff --git a/taskflow/orchestrator/manager_test.go b/taskflow/orchestrator/manager_test.go index 4697474..2f23a93 100644 --- a/taskflow/orchestrator/manager_test.go +++ b/taskflow/orchestrator/manager_test.go @@ -896,3 +896,44 @@ func TestTaskViews_CarryStepIDAndVersion(t *testing.T) { t.Errorf("list = %+v", list) } } + +// The task view and the list carry the claim, and leave it out of the JSON when there is none. +func TestTaskViews_CarryTheClaim(t *testing.T) { + db := newSafeMockTaskStore() + ctx := context.Background() + db.InitTask(ctx, store.TaskRecord{TaskID: "task-c", TaskType: "TEST", State: "PENDING_USER", Data: map[string]any{}}) + tm := newTestTaskManager(db, newTestRegistry(), &mockTemporalManager{}, noopCallback) + + view, err := tm.GetTaskRenderInfo(ctx, "task-c") + if err != nil { + t.Fatal(err) + } + encoded, err := json.Marshal(view) + if err != nil { + t.Fatal(err) + } + var fields map[string]any + if err := json.Unmarshal(encoded, &fields); err != nil { + t.Fatal(err) + } + for _, key := range []string{"claimed_by", "claimed_at"} { + if _, ok := fields[key]; ok { + t.Errorf("unclaimed view has %q: %s", key, encoded) + } + } + + if err := tm.ClaimTask(ctx, "task-c", "user-a"); err != nil { + t.Fatal(err) + } + view, err = tm.GetTaskRenderInfo(ctx, "task-c") + if err != nil { + t.Fatal(err) + } + if view.ClaimedBy != "user-a" || view.ClaimedAt == nil { + t.Errorf("view claim = (%q, %v), want (user-a, a time)", view.ClaimedBy, view.ClaimedAt) + } + list := tm.GetAllTasks(ctx, "") + if len(list) != 1 || list[0].ClaimedBy != "user-a" || list[0].ClaimedAt == nil { + t.Errorf("list = %+v", list) + } +} diff --git a/taskflow/orchestrator/view.go b/taskflow/orchestrator/view.go index 976a417..4ea7f31 100644 --- a/taskflow/orchestrator/view.go +++ b/taskflow/orchestrator/view.go @@ -21,7 +21,11 @@ type TaskView struct { StepID string `json:"step_id,omitempty"` // Version increases every time the task moves on. It is read-only: clients use it to keep the // newest of several views they fetched, and it is never accepted as input. - Version int64 `json:"version"` + Version int64 `json:"version"` + // ClaimedBy is who holds the task's claim, as the host recorded it (see TaskManager.ClaimTask); + // empty when the task is unclaimed. ClaimedAt is when they claimed it. + ClaimedBy string `json:"claimed_by,omitempty"` + ClaimedAt *time.Time `json:"claimed_at,omitempty"` View json.RawMessage `json:"view,omitempty"` CreatedAt time.Time `json:"created_at"` UpdatedAt time.Time `json:"updated_at"` diff --git a/taskflow/renderer/zoneview/zone_assembler.go b/taskflow/renderer/zoneview/zone_assembler.go index e808f9b..4cc2058 100644 --- a/taskflow/renderer/zoneview/zone_assembler.go +++ b/taskflow/renderer/zoneview/zone_assembler.go @@ -45,6 +45,8 @@ func (a *ZoneViewAssembler) Assemble(ctx context.Context, record store.TaskRecor State: record.State, StepID: record.ActiveStepID, Version: record.Seq, + ClaimedBy: record.ClaimedBy, + ClaimedAt: record.ClaimedAt, View: view, CreatedAt: record.CreatedAt, UpdatedAt: record.UpdatedAt, diff --git a/taskflow/renderer/zoneview/zone_assembler_test.go b/taskflow/renderer/zoneview/zone_assembler_test.go index 206ae14..d12af0e 100644 --- a/taskflow/renderer/zoneview/zone_assembler_test.go +++ b/taskflow/renderer/zoneview/zone_assembler_test.go @@ -8,6 +8,7 @@ import ( "encoding/json" "strings" "testing" + "time" "github.com/OpenNSW/core/taskflow/store" ) @@ -170,3 +171,32 @@ func TestAssemble_CarriesStepAndVersionAndOffersNoActionsWhileAdvancing(t *testi } } } + +// The view carries the claim the record holds, and leaves it out of the JSON when there is none. +func TestAssemble_CarriesTheClaim(t *testing.T) { + a := newTestAssembler(t) + claims := map[string]bool{"role:trader": false, "role:cha": true} + + zv, err := a.Assemble(context.Background(), pendingRecord(claimGatedConfig), claims) + if err != nil { + t.Fatalf("Assemble: %v", err) + } + encoded, err := json.Marshal(zv) + if err != nil { + t.Fatal(err) + } + if strings.Contains(string(encoded), `"claimed_by"`) || strings.Contains(string(encoded), `"claimed_at"`) { + t.Errorf("unclaimed view carries a claim: %s", encoded) + } + + at := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) + claimed := pendingRecord(claimGatedConfig) + claimed.ClaimedBy, claimed.ClaimedAt = "user-a", &at + zv, err = a.Assemble(context.Background(), claimed, claims) + if err != nil { + t.Fatalf("Assemble: %v", err) + } + if zv.ClaimedBy != "user-a" || zv.ClaimedAt == nil || !zv.ClaimedAt.Equal(at) { + t.Errorf("view claim = (%q, %v), want (user-a, %v)", zv.ClaimedBy, zv.ClaimedAt, at) + } +} diff --git a/taskflow/renderer/zoneview/zoneview.go b/taskflow/renderer/zoneview/zoneview.go index 81cd5cf..3d77c90 100644 --- a/taskflow/renderer/zoneview/zoneview.go +++ b/taskflow/renderer/zoneview/zoneview.go @@ -94,8 +94,13 @@ type ZoneView struct { State string `json:"state"` // StepID is the step the task is on; actions taken on this view must send it back. Version // increases every time the task moves on; it is read-only and never accepted as input. - StepID string `json:"step_id,omitempty"` - Version int64 `json:"version"` + StepID string `json:"step_id,omitempty"` + Version int64 `json:"version"` + // ClaimedBy is who holds the task's claim, as the host recorded it; empty when the task is + // unclaimed. ClaimedAt is when they claimed it. The view is the same whoever holds the claim: + // to show or hide sections by it, the host passes a claim fact to Assemble. + ClaimedBy string `json:"claimed_by,omitempty"` + ClaimedAt *time.Time `json:"claimed_at,omitempty"` View json.RawMessage `json:"view"` CreatedAt time.Time `json:"created_at"` UpdatedAt time.Time `json:"updated_at"` From 36bb7e0a1ab61dd295bff3438ac90d43c50305d9 Mon Sep 17 00:00:00 2001 From: Janitha Karunarathna Date: Sat, 10 Oct 2026 23:28:27 +0530 Subject: [PATCH 5/8] docs(taskflow): document task claims Describe the claim columns and guarded claim writes for TaskStore implementers, the SQL hosts must add, what taskflow stores versus what the host enforces, the error-to-HTTP mapping, and the claimed_by and claimed_at fields on task views. Distinguish task claims from ClaimStep and from the claim facts passed to the renderer. --- taskflow/README.md | 3 ++ taskflow/docs/architecture.md | 19 ++++++++++ taskflow/docs/frontend-guide.md | 4 +++ taskflow/docs/integration-guide.md | 56 ++++++++++++++++++++++++++++-- 4 files changed, 79 insertions(+), 3 deletions(-) diff --git a/taskflow/README.md b/taskflow/README.md index ec8956d..980a638 100644 --- a/taskflow/README.md +++ b/taskflow/README.md @@ -103,6 +103,8 @@ if err := workflowRunner.StartWorker(); err != nil { | `CompleteTaskStep(ctx, taskID, stepID, payload)` | Portal HTTP handler | Submit form/interaction, resume that step | | `CompleteTaskStepByToken(ctx, token, payload)` | External callback handler | Same, addressed by the dispatch's callback token | | `GetAllTasks(ctx, parentWorkflowID)` | Portal HTTP handler | List tasks for a workflow instance | +| `ClaimTask(ctx, taskID, holder)` | Portal HTTP handler | Claim the task for holder (unclaimed, or already theirs) | +| `ReleaseTask(ctx, taskID, holder)` | Portal HTTP handler | Release holder's claim | ## Writing a plugin @@ -132,6 +134,7 @@ Register it: `pluginRegistry.Register("MY_PLUGIN", &MyPlugin{remoteManager: rm}) - **`TaskID` is the parent's step ID.** A step is one run of a TASK node, so a node revisited by a loop starts a new task. A step is addressed by its step ID (`CompleteTaskStep(ctx, taskID, stepID, payload)`), and a call for a step that is no longer active is rejected with `ErrStaleStep`. - **The task row is written through guarded statements.** `ActiveStepID` and `Seq` change only via `ClaimStep`, `WriteRenderState`, `PersistSubmission` and `CompleteTask`, each one atomic conditional write whose 0-rows result means "stale, dropped". +- **taskflow stores claims; the host enforces them.** `ClaimedBy` and `ClaimedAt` change only via `ClaimTask` and `ReleaseTask`, never via the step writes, so a claim lasts for the whole task. Who may claim, and what a claim is required for, is the host's decision (see [`docs/integration-guide.md`](docs/integration-guide.md#7-task-claims)). - **Plugins suspend with `plugins.ErrSuspended`.** Synchronous plugins return `nil`; the workflow advances without waiting. - **`StartTask` returns `activity.ErrResultPending`** on the happy path. The parent activity suspends until `onTaskCompleted` fires. - **Submission payloads are scoped** to the active step's `OutputNamespace` in `TaskRecord.Data`. Callers send a raw object; the server stamps the correct key. diff --git a/taskflow/docs/architecture.md b/taskflow/docs/architecture.md index e5d25cc..2a317a9 100644 --- a/taskflow/docs/architecture.md +++ b/taskflow/docs/architecture.md @@ -126,6 +126,10 @@ type TaskRecord struct { // 4. Render data — the active step's inputs plus what it and the caller added Data map[string]any + // 5. Claim — who is working the task; "" / nil when unclaimed + ClaimedBy string + ClaimedAt *time.Time + CreatedAt, UpdatedAt time.Time } ``` @@ -152,6 +156,21 @@ Two sets of coordinates, never both active at once: - **Parent coordinates** are written once at `StartTask` and consumed once at `HandleTaskCompletion`. - **The active step** is overwritten every time a new step claims the row, and is what `CompleteTaskStep` checks against. +### Claims + +A claim records who is working a task. `ClaimedBy` and `ClaimedAt` are written only by two more conditional statements, `ClaimTask` and `ReleaseTask`, called by the `TaskManager` methods of the same name: + +| Method | Guard | Sets | +|---------------|---------------------------------------------------------|-----------------------------------------------| +| `ClaimTask` | unclaimed or claimed by the holder, and not `COMPLETED` | holder; claim time unless the holder already had it | +| `ReleaseTask` | claimed by the holder | clears both | + +The claim columns and the step columns are independent: a claim never changes `Seq`, so it does not make an open view stale, and no step write touches the claim. The claim therefore lasts for the whole task, across every step and every loop inside the task workflow, and is kept after the task completes. A loop in the parent workflow starts a new task, which starts unclaimed. + +taskflow only stores claims. The holder is an opaque value chosen by the host, and the host decides who may claim and what a claim is required for (see the [integration guide](integration-guide.md#7-task-claims)). + +These are not the same as a step's claim (`ClaimStep`, above, which makes a step active), or the claims passed to the renderer (`renderer.Facts.Claims`, authorization facts such as `role:officer`). + ## Lifecycle and states `TaskRecord.State` is plain string for flexibility — it's set by plugins and read by the renderer. The orchestrator itself writes these values: diff --git a/taskflow/docs/frontend-guide.md b/taskflow/docs/frontend-guide.md index a71bcd1..e899226 100644 --- a/taskflow/docs/frontend-guide.md +++ b/taskflow/docs/frontend-guide.md @@ -48,6 +48,8 @@ The demo (`demo/server.go`) ships a minimal HTTP surface that maps each `TaskMan "state": "PENDING_USER", "step_id": "0a0a0a0a-0000-4000-8000-00000000000a", "version": 4, + "claimed_by": "4c1d2e3f-0000-4000-8000-000000000001", + "claimed_at": "2026-05-18T09:20:02Z", "created_at": "2026-05-18T09:14:11Z", "updated_at": "2026-05-18T09:14:11Z" } @@ -80,6 +82,8 @@ The `view` key is omitted (the JSON tag is `omitempty`). Listing is intentionall `step_id` is the step the task is on; send it back when you act on the view. `version` goes up every time the task moves on. It is read-only — never send it — and is there so you can keep the newest of several views you fetched. +`claimed_by` and `claimed_at` say who is working the task and since when. Both are left out when nobody has claimed it. `claimed_by` is whatever your server recorded, usually an internal user ID, so don't show it as is or compare it with values from your login token. Your server decides what a claim is required for, and will normally add a display name and whether the claim is yours. Claiming does not change `version`. + ### `View` JSON structure The `view` field returned in the API is a raw JSON payload (`json.RawMessage`). The core orchestrator does not enforce any specific schema, allowing you to use whatever layout paradigm fits your frontend. diff --git a/taskflow/docs/integration-guide.md b/taskflow/docs/integration-guide.md index a2acb77..e6a9dfa 100644 --- a/taskflow/docs/integration-guide.md +++ b/taskflow/docs/integration-guide.md @@ -43,6 +43,10 @@ type TaskStore interface { WriteRenderState(ctx context.Context, taskID, stepID string, seq int64, state string, data map[string]any) (int64, error) PersistSubmission(ctx context.Context, taskID, stepID string, seq int64, data map[string]any) (int64, error) CompleteTask(ctx context.Context, taskID string, seq int64) (int64, error) + + // Guarded claim writes: the same kind of statement, on the claim columns only. + ClaimTask(ctx context.Context, taskID, holder string, at time.Time) (int64, error) + ReleaseTask(ctx context.Context, taskID, holder string) (int64, error) } ``` @@ -62,14 +66,26 @@ type TaskStore interface { | `CompleteTask` | stored `seq <= seq` | `state = COMPLETED`, `seq` | Return the number of rows changed. **0 rows means the write was stale and was dropped; it is not an error.** Return an error only when the store failed. +- **The two claim methods** are conditional writes of the same kind, on `claimed_by` and `claimed_at` only. They must not change `seq` or the step columns, and the step methods and `InitTask` must not change the claim columns: + + | Method | Guard | Sets | + |---------------|--------------------------------------------------------------------|------------------------------------------------------------------------------| + | `ClaimTask` | (`claimed_by` is NULL or `holder`) and `state` is not `COMPLETED` | `claimed_by = holder`; `claimed_at = at`, unless `holder` already had the claim | + | `ReleaseTask` | `claimed_by = holder` | `claimed_by` and `claimed_at` to NULL | + + Store an unclaimed task as NULL, not `""`, so the guard's `IS NULL` matches it. See [Task claims](#7-task-claims) for what the manager builds on these. A GORM implementation is in `store/gorm`. Its table needs, beyond the original columns: ```sql ALTER TABLE task_records_v2 ADD COLUMN IF NOT EXISTS active_step_id UUID NULL; ALTER TABLE task_records_v2 ADD COLUMN IF NOT EXISTS seq BIGINT NOT NULL DEFAULT 0; +ALTER TABLE task_records_v2 ADD COLUMN IF NOT EXISTS claimed_by TEXT NULL; +ALTER TABLE task_records_v2 ADD COLUMN IF NOT EXISTS claimed_at TIMESTAMPTZ NULL; ``` +`InitTask` leaves the claim columns out of its insert, so tasks can still be created before they exist. Add them before anything calls `ClaimTask` or `ReleaseTask`. + ### Concurrency The store is called from multiple goroutines (Temporal worker pool, HTTP handlers). Your implementation must be safe for concurrent use. The demo uses a `sync.RWMutex`; a SQL-backed store gets this from the DB. The conditional writes are what make racing callers, retried activities and late attempts safe, so they must really be atomic. @@ -229,9 +245,43 @@ The render config is snapshotted into the `TaskRecord` at `StartTask` time. Once --- +## 7. Task claims + +A claim records who is working a task, so two people don't act on it at once and everyone can see who has it. The split of work is: + +- **taskflow stores claims.** `TaskManager.ClaimTask(ctx, taskID, holder)` and `ReleaseTask(ctx, taskID, holder)` are atomic, and task views carry `claimed_by` and `claimed_at`. taskflow never decides who may claim a task or what a claim is required for. +- **Your host enforces them.** It exposes claim and release endpoints, chooses the holder value, and decides which actions need the claim. + +The rules taskflow applies: + +- A claim succeeds when the task is unclaimed or already claimed by `holder`. A repeat claim keeps the original `claimed_at`. +- A completed task cannot be claimed. A claim held when it completed is kept, as a record of who worked it, and can still be released. +- Only the holder can release. Releasing an unclaimed task succeeds, so a repeated release is harmless. +- A claim never changes the task's step or version, so it does not make an open view stale. +- The claim is on the task row, so it lasts for the whole task, across every step inside it. A loop **inside** the task workflow (for example, sending a form back for more information) keeps the claim. A loop in the **parent** workflow starts a new task, which starts unclaimed. + +What your host needs to do: + +1. **Choose the holder value.** Use a stable, internal user ID, not an email or a display name. taskflow treats it as opaque. Resolve it to a name when you build responses. +2. **Expose claim and release** for callers who can see the task, and map the errors (below). +3. **Enforce the claim where your policy needs it.** For example, before calling `CompleteTaskStep`, reject a user who does not hold the claim. Exempt machine callers such as `CompleteTaskStepByToken`. The check and the submit are two calls, but that gap is harmless: the workflow still accepts only one submission for a step. +4. **Optionally gate the view on the claim.** Pass a fact such as `claim:mine` in the claims you give the renderer, and use it in `visibleWhen.requireClaim` to hide actions from someone who does not hold the claim. The renderer rejects a render config that references a fact you did not pass, so pass every claim fact your configs use on every read, with `false` for the ones that don't hold. + +| Error | Returned by | HTTP | +|----------------------------|--------------------------------|------| +| `ErrHolderRequired` | `ClaimTask`, `ReleaseTask` | 400 | +| `ErrTaskNotFound` | `ClaimTask`, `ReleaseTask` | 404 | +| `*ClaimHeldError` (matches `ErrClaimHeld`) | `ClaimTask` | 409; its `Holder` says who has the claim | +| `ErrTaskCompleted` | `ClaimTask` | 409 | +| `ErrNotClaimHolder` | `ReleaseTask` | 403 | + +"Claim" here is not `TaskStore.ClaimStep`, which makes a step the task's active step, and not the claims passed to the renderer (`renderer.Facts.Claims`), which are authorization facts such as `role:officer`. + +--- + ## Error semantics -`ErrStaleStep` and `ErrStepIDRequired` (see `CompleteTaskStep`) are for your HTTP layer to map to 409 and 400. `engine.ErrActivationNotPending` is what `CompleteActivation` returns for a step that already completed or never existed. +`ErrStaleStep` and `ErrStepIDRequired` (see `CompleteTaskStep`) are for your HTTP layer to map to 409 and 400. `ErrTaskNotFound`, from `CompleteTaskStep`, `GetTaskRenderInfo` and the claim methods, maps to 404. The claim errors are listed in [Task claims](#7-task-claims). `engine.ErrActivationNotPending` is what `CompleteActivation` returns for a step that already completed or never existed. The orchestrator uses two sentinel errors to signal "this isn't really an error, the workflow should park": @@ -314,9 +364,9 @@ Things the demo cuts corners on, that you'll want for a real deployment: - **Durable `TaskStore`.** The demo writes JSON to `/tmp`. Use Postgres / Spanner / DynamoDB. - **Idempotent `onTaskCompleted`.** Temporal retries; you must tolerate replay. - **Observability.** Plugins log via `log.Printf`. Replace with your structured logger; surface task state transitions, plugin errors, and callback failures to your observability stack. -- **Schema.** Add `active_step_id` and `seq` to your task table (section 1) before deploying a version that uses the guarded writes. +- **Schema.** Add `active_step_id` and `seq` to your task table (section 1) before deploying a version that uses the guarded writes, and `claimed_by` and `claimed_at` before using task claims. - **Plugin error handling.** Decide what should retry vs. fail-fast. The orchestrator currently surfaces non-`ErrSuspended` plugin errors to Temporal, which will retry per the activity's retry policy. -- **Authorisation.** `CompleteTaskStep` accepts any payload for any known task. Add authn/authz at your HTTP layer. +- **Authorisation.** `CompleteTaskStep` accepts any payload for any known task. Add authn/authz at your HTTP layer, including any claim requirement (section 7). --- From efa02bb54ec977df00d87a65ef3ae5a4f800b2e5 Mon Sep 17 00:00:00 2001 From: Janitha Karunarathna Date: Sun, 11 Oct 2026 23:42:09 +0530 Subject: [PATCH 6/8] docs(taskflow): say how clients take claim state when version does not change Claiming and releasing leave version unchanged, so a client that keeps the newest view by version cannot tell a claim change from no change. Tell clients to take claim fields from the most recent response instead, and to refetch after their own claim or release. Note that out-of-order responses can still hide a claim change, and that a revision would fix it. --- taskflow/docs/frontend-guide.md | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/taskflow/docs/frontend-guide.md b/taskflow/docs/frontend-guide.md index e899226..9b68357 100644 --- a/taskflow/docs/frontend-guide.md +++ b/taskflow/docs/frontend-guide.md @@ -82,7 +82,9 @@ The `view` key is omitted (the JSON tag is `omitempty`). Listing is intentionall `step_id` is the step the task is on; send it back when you act on the view. `version` goes up every time the task moves on. It is read-only — never send it — and is there so you can keep the newest of several views you fetched. -`claimed_by` and `claimed_at` say who is working the task and since when. Both are left out when nobody has claimed it. `claimed_by` is whatever your server recorded, usually an internal user ID, so don't show it as is or compare it with values from your login token. Your server decides what a claim is required for, and will normally add a display name and whether the claim is yours. Claiming does not change `version`. +`claimed_by` and `claimed_at` say who is working the task and since when. Both are left out when nobody has claimed it. `claimed_by` is whatever your server recorded, usually an internal user ID, so don't show it as is or compare it with values from your login token. Your server decides what a claim is required for, and will normally add a display name and whether the claim is yours. + +Claiming and releasing do not change `version`, so `version` cannot tell you whether the claim changed. Take the claim fields from the most recent response you received, not from the one with the higher `version`, and refetch the task after your own claim or release. If responses for the same task can arrive out of order, a stale response with the same `version` can still overwrite a newer claim. Treat that as a known limit: the robust fix is a revision that changes on every claim change, which core does not expose yet. ### `View` JSON structure From 1ed684552e52cc4a230b5ec5fb4e63587bb87fbb Mon Sep 17 00:00:00 2001 From: Janitha Karunarathna Date: Sun, 11 Oct 2026 23:43:55 +0530 Subject: [PATCH 7/8] docs(taskflow): fence claim checks with the submission A check made before CompleteTaskStep is not safe on its own: between the check and the submit the holder can release, and another caller can take the claim. The workflow does not know who holds the claim, so it accepts the submission from whoever made the call, and the one-submission rule does not stop that. Say so, and say that claim-protected submissions must serialise the check and the submit per task, with release and claim changes waiting on the same lock. The machine-caller exemption on the token route is unchanged. --- taskflow/docs/integration-guide.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/taskflow/docs/integration-guide.md b/taskflow/docs/integration-guide.md index e6a9dfa..20ec6ec 100644 --- a/taskflow/docs/integration-guide.md +++ b/taskflow/docs/integration-guide.md @@ -264,7 +264,7 @@ What your host needs to do: 1. **Choose the holder value.** Use a stable, internal user ID, not an email or a display name. taskflow treats it as opaque. Resolve it to a name when you build responses. 2. **Expose claim and release** for callers who can see the task, and map the errors (below). -3. **Enforce the claim where your policy needs it.** For example, before calling `CompleteTaskStep`, reject a user who does not hold the claim. Exempt machine callers such as `CompleteTaskStepByToken`. The check and the submit are two calls, but that gap is harmless: the workflow still accepts only one submission for a step. +3. **Enforce the claim where your policy needs it, fenced with the submission.** Exempt machine callers such as `CompleteTaskStepByToken`, which hold no claim. A separate check before `CompleteTaskStep` is not safe on its own. Between the check and the submit, the holder can release the claim and another caller can take it. The workflow does not know who holds the claim: `CompleteActivation` accepts the submission from whoever made the call. The one-submission rule only stops a step being completed twice; it does not stop a submission from someone who no longer holds the claim. For claim-protected actions, serialise the check and the submit per task, for example by taking a row lock on the task across both, or by re-checking the holder inside the same lock that the submit runs under. Release and claim changes must wait for that lock too, so they cannot land between the check and the submit. 4. **Optionally gate the view on the claim.** Pass a fact such as `claim:mine` in the claims you give the renderer, and use it in `visibleWhen.requireClaim` to hide actions from someone who does not hold the claim. The renderer rejects a render config that references a fact you did not pass, so pass every claim fact your configs use on every read, with `false` for the ones that don't hold. | Error | Returned by | HTTP | From e4d09d03c0f7a0893dcad4439a1b8568564ff34a Mon Sep 17 00:00:00 2001 From: Janitha Karunarathna Date: Tue, 13 Oct 2026 08:44:18 +0530 Subject: [PATCH 8/8] docs(taskflow): describe claims as UI guidance CompleteTaskStep accepts a submission from anyone, so document a claim as guidance that shows who is working a task instead of a lock around submission. Co-authored-by: Cursor --- taskflow/README.md | 2 +- taskflow/docs/architecture.md | 2 +- taskflow/docs/frontend-guide.md | 2 +- taskflow/docs/integration-guide.md | 12 ++++++------ taskflow/orchestrator/claim.go | 8 ++++---- 5 files changed, 13 insertions(+), 13 deletions(-) diff --git a/taskflow/README.md b/taskflow/README.md index 980a638..a44c0c7 100644 --- a/taskflow/README.md +++ b/taskflow/README.md @@ -134,7 +134,7 @@ Register it: `pluginRegistry.Register("MY_PLUGIN", &MyPlugin{remoteManager: rm}) - **`TaskID` is the parent's step ID.** A step is one run of a TASK node, so a node revisited by a loop starts a new task. A step is addressed by its step ID (`CompleteTaskStep(ctx, taskID, stepID, payload)`), and a call for a step that is no longer active is rejected with `ErrStaleStep`. - **The task row is written through guarded statements.** `ActiveStepID` and `Seq` change only via `ClaimStep`, `WriteRenderState`, `PersistSubmission` and `CompleteTask`, each one atomic conditional write whose 0-rows result means "stale, dropped". -- **taskflow stores claims; the host enforces them.** `ClaimedBy` and `ClaimedAt` change only via `ClaimTask` and `ReleaseTask`, never via the step writes, so a claim lasts for the whole task. Who may claim, and what a claim is required for, is the host's decision (see [`docs/integration-guide.md`](docs/integration-guide.md#7-task-claims)). +- **taskflow stores claims; `CompleteTaskStep` does not check them.** `ClaimedBy` and `ClaimedAt` change only via `ClaimTask` and `ReleaseTask`, never via the step writes, so a claim lasts for the whole task. For this MVP a claim is UI guidance so two people can avoid working the same task (see [`docs/integration-guide.md`](docs/integration-guide.md#7-task-claims)). - **Plugins suspend with `plugins.ErrSuspended`.** Synchronous plugins return `nil`; the workflow advances without waiting. - **`StartTask` returns `activity.ErrResultPending`** on the happy path. The parent activity suspends until `onTaskCompleted` fires. - **Submission payloads are scoped** to the active step's `OutputNamespace` in `TaskRecord.Data`. Callers send a raw object; the server stamps the correct key. diff --git a/taskflow/docs/architecture.md b/taskflow/docs/architecture.md index 2a317a9..fca03c5 100644 --- a/taskflow/docs/architecture.md +++ b/taskflow/docs/architecture.md @@ -167,7 +167,7 @@ A claim records who is working a task. `ClaimedBy` and `ClaimedAt` are written o The claim columns and the step columns are independent: a claim never changes `Seq`, so it does not make an open view stale, and no step write touches the claim. The claim therefore lasts for the whole task, across every step and every loop inside the task workflow, and is kept after the task completes. A loop in the parent workflow starts a new task, which starts unclaimed. -taskflow only stores claims. The holder is an opaque value chosen by the host, and the host decides who may claim and what a claim is required for (see the [integration guide](integration-guide.md#7-task-claims)). +taskflow only stores claims. The holder is an opaque value chosen by the host. `CompleteTaskStep` does not read `ClaimedBy`: anyone who can call it can submit the active step. For this MVP a claim is UI guidance so two people can see who is working the task and avoid working it at once (see the [integration guide](integration-guide.md#7-task-claims)). These are not the same as a step's claim (`ClaimStep`, above, which makes a step active), or the claims passed to the renderer (`renderer.Facts.Claims`, authorization facts such as `role:officer`). diff --git a/taskflow/docs/frontend-guide.md b/taskflow/docs/frontend-guide.md index 9b68357..a769336 100644 --- a/taskflow/docs/frontend-guide.md +++ b/taskflow/docs/frontend-guide.md @@ -82,7 +82,7 @@ The `view` key is omitted (the JSON tag is `omitempty`). Listing is intentionall `step_id` is the step the task is on; send it back when you act on the view. `version` goes up every time the task moves on. It is read-only — never send it — and is there so you can keep the newest of several views you fetched. -`claimed_by` and `claimed_at` say who is working the task and since when. Both are left out when nobody has claimed it. `claimed_by` is whatever your server recorded, usually an internal user ID, so don't show it as is or compare it with values from your login token. Your server decides what a claim is required for, and will normally add a display name and whether the claim is yours. +`claimed_by` and `claimed_at` say who is working the task and since when. Both are left out when nobody has claimed it. `claimed_by` is whatever your server recorded, usually an internal user ID, so don't show it as is or compare it with values from your login token. Show them so two people can avoid working the same task. Submitting a step does not require the claim. Your server will normally add a display name and whether the claim is yours. Claiming and releasing do not change `version`, so `version` cannot tell you whether the claim changed. Take the claim fields from the most recent response you received, not from the one with the higher `version`, and refetch the task after your own claim or release. If responses for the same task can arrive out of order, a stale response with the same `version` can still overwrite a newer claim. Treat that as a known limit: the robust fix is a revision that changes on every claim change, which core does not expose yet. diff --git a/taskflow/docs/integration-guide.md b/taskflow/docs/integration-guide.md index 20ec6ec..486bc57 100644 --- a/taskflow/docs/integration-guide.md +++ b/taskflow/docs/integration-guide.md @@ -247,10 +247,10 @@ The render config is snapshotted into the `TaskRecord` at `StartTask` time. Once ## 7. Task claims -A claim records who is working a task, so two people don't act on it at once and everyone can see who has it. The split of work is: +A claim records who is working a task, so the UI can show it and two people can avoid working the same task at once. For this MVP it is guidance. `CompleteTaskStep` does not read `claimed_by`: anyone who can call it can submit the active step. The split of work is: -- **taskflow stores claims.** `TaskManager.ClaimTask(ctx, taskID, holder)` and `ReleaseTask(ctx, taskID, holder)` are atomic, and task views carry `claimed_by` and `claimed_at`. taskflow never decides who may claim a task or what a claim is required for. -- **Your host enforces them.** It exposes claim and release endpoints, chooses the holder value, and decides which actions need the claim. +- **taskflow stores claims.** `TaskManager.ClaimTask(ctx, taskID, holder)` and `ReleaseTask(ctx, taskID, holder)` are atomic, and task views carry `claimed_by` and `claimed_at`. Submitting a step does not check the holder. +- **Your host shows them.** It exposes claim and release endpoints, chooses the holder value, and shows who holds the claim. Submitting a step stays available to anyone who can call it. The rules taskflow applies: @@ -264,8 +264,8 @@ What your host needs to do: 1. **Choose the holder value.** Use a stable, internal user ID, not an email or a display name. taskflow treats it as opaque. Resolve it to a name when you build responses. 2. **Expose claim and release** for callers who can see the task, and map the errors (below). -3. **Enforce the claim where your policy needs it, fenced with the submission.** Exempt machine callers such as `CompleteTaskStepByToken`, which hold no claim. A separate check before `CompleteTaskStep` is not safe on its own. Between the check and the submit, the holder can release the claim and another caller can take it. The workflow does not know who holds the claim: `CompleteActivation` accepts the submission from whoever made the call. The one-submission rule only stops a step being completed twice; it does not stop a submission from someone who no longer holds the claim. For claim-protected actions, serialise the check and the submit per task, for example by taking a row lock on the task across both, or by re-checking the holder inside the same lock that the submit runs under. Release and claim changes must wait for that lock too, so they cannot land between the check and the submit. -4. **Optionally gate the view on the claim.** Pass a fact such as `claim:mine` in the claims you give the renderer, and use it in `visibleWhen.requireClaim` to hide actions from someone who does not hold the claim. The renderer rejects a render config that references a fact you did not pass, so pass every claim fact your configs use on every read, with `false` for the ones that don't hold. +3. **Show the claim in the UI.** `CompleteTaskStep` accepts a call from anyone and does not read `claimed_by`. For this MVP, a claim is guidance so two people can see who is working the task and avoid working it at once. Machine callers such as `CompleteTaskStepByToken` hold no claim and can still complete the step. +4. **Optionally hide actions in the view.** Pass a fact such as `claim:mine` in the claims you give the renderer, and use it in `visibleWhen.requireClaim` to hide actions from someone who does not hold the claim. That only changes what the view shows. The submit call is still accepted. The renderer rejects a render config that references a fact you did not pass, so pass every claim fact your configs use on every read, with `false` for the ones that don't hold. | Error | Returned by | HTTP | |----------------------------|--------------------------------|------| @@ -366,7 +366,7 @@ Things the demo cuts corners on, that you'll want for a real deployment: - **Observability.** Plugins log via `log.Printf`. Replace with your structured logger; surface task state transitions, plugin errors, and callback failures to your observability stack. - **Schema.** Add `active_step_id` and `seq` to your task table (section 1) before deploying a version that uses the guarded writes, and `claimed_by` and `claimed_at` before using task claims. - **Plugin error handling.** Decide what should retry vs. fail-fast. The orchestrator currently surfaces non-`ErrSuspended` plugin errors to Temporal, which will retry per the activity's retry policy. -- **Authorisation.** `CompleteTaskStep` accepts any payload for any known task. Add authn/authz at your HTTP layer, including any claim requirement (section 7). +- **Authorisation.** `CompleteTaskStep` accepts any payload for any known task. Add authn/authz at your HTTP layer. A task claim is not part of that check (see [Task claims](#7-task-claims)). --- diff --git a/taskflow/orchestrator/claim.go b/taskflow/orchestrator/claim.go index 3f803af..c06927a 100644 --- a/taskflow/orchestrator/claim.go +++ b/taskflow/orchestrator/claim.go @@ -12,10 +12,10 @@ import ( "github.com/OpenNSW/core/taskflow/store" ) -// A claim records who is working a task, so two people do not act on it at once. The holder is an -// opaque value chosen by the host (for example an internal user ID). taskflow only stores claims: -// it never decides who may claim a task or what a claim is required for. That is the host's -// policy, enforced in its own handlers before it calls CompleteTaskStep. +// A claim records who is working a task, so the UI can show it and two people can avoid working +// the same task at once. The holder is an opaque value chosen by the host (for example an internal +// user ID). taskflow only stores claims. CompleteTaskStep does not read the holder: anyone who can +// call it can submit the active step. For this MVP a claim is UI guidance. // // Not to be confused with store.TaskStore.ClaimStep, which makes a step the task's active step, or // with renderer.Facts.Claims, which are authorization facts for rendering.