feat(taskflow)!: claim and release tasks

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; `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.

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. `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`).
+
 ## 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:

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,10 @@ 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. 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.
+
 ### `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.

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 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`. 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:
+
+- 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. **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 |
+|----------------------------|--------------------------------|------|
+| `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. A task claim is not part of that check (see [Task claims](#7-task-claims)).
 
 ---
 

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 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.
+
+// 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)
+}