diff --git a/internal/workitems/inputs.go b/internal/workitems/inputs.go new file mode 100644 index 00000000..4c425e42 --- /dev/null +++ b/internal/workitems/inputs.go @@ -0,0 +1,310 @@ +package workitems + +import ( + "encoding/json" + "fmt" + "strings" + "unicode/utf8" + + "github.com/mctlhq/mctl-api/internal/secretscan" +) + +// Mutation is what every write carries besides its own fields. Actor is the +// authenticated principal the HTTP layer derived; the store records it and +// never looks for one anywhere else. +type Mutation struct { + Actor string + Surface string + RequestID string + // IdempotencyKey deduplicates the request: per tenant for Create, per + // work item for everything else. RequestHash is the digest of the + // request the key was first used with. The same key with a different + // actor or request hash is ErrIdempotencyKeyReuse, never a silent + // replay. + IdempotencyKey string + RequestHash string +} + +// CreateInput opens a work item in state active. +type CreateInput struct { + Mutation + Tenant string + Visibility string + OriginSurface string + Title string + // ExternalKey (for example a GitHub issue URL) dedupes open work: while + // a non-terminal item in the tenant carries it, Create returns that item. + ExternalKey string +} + +// TransitionInput moves a work item along Transitions. +type TransitionInput struct { + Mutation + WorkItemID string + Action string + WaitingReason string // required for wait, forbidden otherwise + SupersededBy string // required for supersede, forbidden otherwise + ExpectedStateVersion int64 +} + +// ResumeInput starts a new execution continuing a prior one. A waiting item +// moves back to active; either way a resumed event is recorded. An item that +// never had an execution can still be resumed (a request parked for input +// before any run started): the new execution is attempt 1 with no +// resumed_from, and Resume stays the only way out of waiting. An active item +// with no execution starts its first run with AttachExecution instead. +type ResumeInput struct { + Mutation + WorkItemID string + ExpectedStateVersion int64 + // ResumedFromExecutionID names the execution being continued. Empty + // means the latest one, if the item has any. + ResumedFromExecutionID string + Engine string + EngineRef string +} + +// IntentInput appends a bounded intent. +type IntentInput struct { + Mutation + WorkItemID string + Text string + Params json.RawMessage +} + +// ExecutionInput attaches an execution, or correlates a later phase of one +// already attached under the same (engine, engine_ref). +type ExecutionInput struct { + Mutation + WorkItemID string + Engine string + EngineRef string + Phase string +} + +// SurfaceRefInput correlates a surface-native conversation. +type SurfaceRefInput struct { + Mutation + WorkItemID string + ExternalID string + ActorExternalID string +} + +// ListFilter selects work items. Tenancy fails closed: an empty Tenants +// matches nothing unless AllTenants is set (admins only; the HTTP layer +// decides). Named Tenants always narrow, AllTenants or not. Viewer, when set, hides +// private items owned by anyone else. +type ListFilter struct { + Tenants []string + AllTenants bool + State string // "" or FilterOpen: non-terminal; otherwise one state + Owner string + Viewer string + Limit int +} + +const ( + defaultListLimit = 100 + maxListLimit = 500 +) + +func invalid(format string, args ...any) error { + return fmt.Errorf("%w: %s", ErrInvalid, fmt.Sprintf(format, args...)) +} + +func checkText(field, value string, max int, required bool) error { + if required && strings.TrimSpace(value) == "" { + return invalid("%s is required", field) + } + if len(value) > max { + return invalid("%s exceeds %d bytes", field, max) + } + if !utf8.ValidString(value) { + return invalid("%s is not valid UTF-8", field) + } + return nil +} + +// checkIdentity is checkText for values matched exactly later (principals, +// tenants): stored as given, padding would make them miss every match, a +// private item invisible to its own owner. +func checkIdentity(field, value string) error { + if err := checkText(field, value, MaxExternalIDBytes, true); err != nil { + return err + } + if value != strings.TrimSpace(value) { + return invalid("%s has leading or trailing whitespace", field) + } + return nil +} + +func (m Mutation) validate() error { + if err := checkIdentity("actor", m.Actor); err != nil { + return err + } + if err := checkText("request_id", m.RequestID, MaxExternalIDBytes, false); err != nil { + return err + } + if m.Surface != "" && !surfacePattern.MatchString(m.Surface) { + return invalid("surface must match %s", surfacePattern) + } + if err := checkText("idempotency_key", m.IdempotencyKey, MaxKeyBytes, false); err != nil { + return err + } + if m.IdempotencyKey != "" && m.RequestHash == "" { + return invalid("an idempotency key needs the request hash it was used with") + } + return nil +} + +func (in CreateInput) validate() error { + if err := in.Mutation.validate(); err != nil { + return err + } + if err := checkIdentity("tenant", in.Tenant); err != nil { + return err + } + if in.Visibility != VisibilityTenant && in.Visibility != VisibilityPrivate { + return invalid("visibility must be %q or %q", VisibilityTenant, VisibilityPrivate) + } + if !surfacePattern.MatchString(in.OriginSurface) { + return invalid("origin_surface must match %s", surfacePattern) + } + if err := checkText("title", in.Title, MaxTitleBytes, true); err != nil { + return err + } + if err := checkText("external_key", in.ExternalKey, MaxKeyBytes, false); err != nil { + return err + } + if err := scanSecrets("title", in.Title); err != nil { + return err + } + return scanSecrets("external_key", in.ExternalKey) +} + +func (in TransitionInput) validate() error { + if err := in.Mutation.validate(); err != nil { + return err + } + if in.WorkItemID == "" { + return invalid("work item id is required") + } + switch in.Action { + case ActionWait, ActionComplete, ActionSupersede, ActionArchive: + case ActionResume: + // Resume starts an execution and has its own route. + return invalid("resume is its own operation") + default: + return invalid("unknown action %q", in.Action) + } + if in.Action == ActionWait { + if in.WaitingReason != WaitingInput && in.WaitingReason != WaitingApproval { + return invalid("wait needs waiting_reason %q or %q", WaitingInput, WaitingApproval) + } + } else if in.WaitingReason != "" { + return invalid("waiting_reason is only valid with wait") + } + if in.Action == ActionSupersede { + if in.SupersededBy == "" || in.SupersededBy == in.WorkItemID { + return invalid("supersede needs superseded_by naming another work item") + } + } else if in.SupersededBy != "" { + return invalid("superseded_by is only valid with supersede") + } + if in.ExpectedStateVersion <= 0 { + return invalid("expected_state_version is required") + } + return nil +} + +func (in ResumeInput) validate() error { + if err := in.Mutation.validate(); err != nil { + return err + } + if in.WorkItemID == "" { + return invalid("work item id is required") + } + if in.ExpectedStateVersion <= 0 { + return invalid("expected_state_version is required") + } + return validateEngine(in.Engine, in.EngineRef) +} + +func (in IntentInput) validate() error { + if err := in.Mutation.validate(); err != nil { + return err + } + if in.WorkItemID == "" { + return invalid("work item id is required") + } + if err := checkText("text", in.Text, MaxIntentBytes, true); err != nil { + return err + } + if len(in.Params) > 0 { + if len(in.Params) > MaxIntentParamBytes { + return invalid("params exceed %d bytes", MaxIntentParamBytes) + } + var obj map[string]any + if err := json.Unmarshal(in.Params, &obj); err != nil || obj == nil { + return invalid("params must be a JSON object") + } + } + if err := scanSecrets("text", in.Text); err != nil { + return err + } + return scanSecrets("params", string(in.Params)) +} + +func (in ExecutionInput) validate() error { + if err := in.Mutation.validate(); err != nil { + return err + } + if in.WorkItemID == "" { + return invalid("work item id is required") + } + if !validPhase(in.Phase) { + return invalid("phase %q is not one of Pending, Running, Succeeded, Failed, Error", in.Phase) + } + return validateEngine(in.Engine, in.EngineRef) +} + +func (in SurfaceRefInput) validate() error { + if err := in.Mutation.validate(); err != nil { + return err + } + if in.WorkItemID == "" { + return invalid("work item id is required") + } + if in.Surface == "" { + return invalid("surface is required") + } + if err := checkText("external_id", in.ExternalID, MaxExternalIDBytes, true); err != nil { + return err + } + if err := checkText("actor_external_id", in.ActorExternalID, MaxExternalIDBytes, false); err != nil { + return err + } + if err := scanSecrets("external_id", in.ExternalID); err != nil { + return err + } + return scanSecrets("actor_external_id", in.ActorExternalID) +} + +func validateEngine(engine, ref string) error { + if engine != EngineTemporal && engine != EngineArgo { + return invalid("engine must be %q or %q", EngineTemporal, EngineArgo) + } + return checkText("engine_ref", ref, MaxEngineRefBytes, true) +} + +// scanSecrets is the contract's gate on free text: a match is rejected and +// never persisted. +func scanSecrets(field, value string) error { + if value == "" { + return nil + } + if pattern, found := secretscan.Scan([]byte(value)); found { + return fmt.Errorf("%w: %s (%s)", ErrSecretInText, field, pattern) + } + return nil +} diff --git a/internal/workitems/store.go b/internal/workitems/store.go new file mode 100644 index 00000000..daf1d275 --- /dev/null +++ b/internal/workitems/store.go @@ -0,0 +1,891 @@ +package workitems + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "log/slog" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +const schema = ` +CREATE TABLE IF NOT EXISTS work_items ( + id TEXT PRIMARY KEY, + tenant TEXT NOT NULL, + owner_principal TEXT NOT NULL, + visibility TEXT NOT NULL, + origin_surface TEXT NOT NULL, + title TEXT NOT NULL, + external_key TEXT NOT NULL DEFAULT '', + state TEXT NOT NULL, + waiting_reason TEXT NOT NULL DEFAULT '', + superseded_by TEXT NOT NULL DEFAULT '', + state_version BIGINT NOT NULL, + created_by TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL, + updated_at TIMESTAMPTZ NOT NULL, + completed_at TIMESTAMPTZ, + schema_version TEXT NOT NULL +); +-- Open-work dedupe: one non-terminal item per (tenant, external_key), the +-- same shape as alerts_tenant_fingerprint_open. +CREATE UNIQUE INDEX IF NOT EXISTS work_items_tenant_external_key_open + ON work_items (tenant, external_key) + WHERE external_key <> '' AND state IN ('active', 'waiting'); +CREATE INDEX IF NOT EXISTS work_items_tenant_state + ON work_items (tenant, state, updated_at DESC); + +CREATE TABLE IF NOT EXISTS work_item_events ( + work_item_id TEXT NOT NULL REFERENCES work_items (id) ON DELETE CASCADE, + seq BIGINT NOT NULL, + kind TEXT NOT NULL, + from_state TEXT NOT NULL DEFAULT '', + to_state TEXT NOT NULL DEFAULT '', + actor_principal TEXT NOT NULL, + surface TEXT NOT NULL DEFAULT '', + request_id TEXT NOT NULL DEFAULT '', + detail JSONB, + created_at TIMESTAMPTZ NOT NULL, + PRIMARY KEY (work_item_id, seq) +); + +-- text is nullable so a retention sweep can drop it and keep the row. +CREATE TABLE IF NOT EXISTS work_item_intents ( + id BIGSERIAL PRIMARY KEY, + work_item_id TEXT NOT NULL REFERENCES work_items (id) ON DELETE CASCADE, + actor_principal TEXT NOT NULL, + surface TEXT NOT NULL DEFAULT '', + text TEXT, + params JSONB, + created_at TIMESTAMPTZ NOT NULL +); +CREATE INDEX IF NOT EXISTS work_item_intents_item ON work_item_intents (work_item_id, id); + +CREATE TABLE IF NOT EXISTS work_item_executions ( + id TEXT PRIMARY KEY, + work_item_id TEXT NOT NULL REFERENCES work_items (id) ON DELETE CASCADE, + engine TEXT NOT NULL, + engine_ref TEXT NOT NULL, + attempt INTEGER NOT NULL, + resumed_from_execution_id TEXT NOT NULL DEFAULT '', + phase TEXT NOT NULL, + started_at TIMESTAMPTZ NOT NULL, + ended_at TIMESTAMPTZ, + UNIQUE (work_item_id, engine, engine_ref), + UNIQUE (work_item_id, attempt) +); +-- At most one non-terminal execution per work item. +CREATE UNIQUE INDEX IF NOT EXISTS work_item_executions_one_open + ON work_item_executions (work_item_id) + WHERE phase IN ('Pending', 'Running'); + +CREATE TABLE IF NOT EXISTS work_item_surface_refs ( + work_item_id TEXT NOT NULL REFERENCES work_items (id) ON DELETE CASCADE, + surface TEXT NOT NULL, + external_id TEXT NOT NULL, + actor_external_id TEXT NOT NULL DEFAULT '', + first_seen_at TIMESTAMPTZ NOT NULL, + last_seen_at TIMESTAMPTZ NOT NULL, + PRIMARY KEY (work_item_id, surface, external_id) +); + +-- Create idempotency per (tenant, key). Kept apart from work_items so a key +-- that deduped onto an existing item (external_key) is recorded too. +CREATE TABLE IF NOT EXISTS work_item_create_requests ( + tenant TEXT NOT NULL, + idempotency_key TEXT NOT NULL, + request_hash TEXT NOT NULL, + actor TEXT NOT NULL, + work_item_id TEXT NOT NULL REFERENCES work_items (id) ON DELETE CASCADE, + created_at TIMESTAMPTZ NOT NULL, + PRIMARY KEY (tenant, idempotency_key) +); + +-- Per-item idempotency for every mutation other than create. +CREATE TABLE IF NOT EXISTS work_item_requests ( + work_item_id TEXT NOT NULL REFERENCES work_items (id) ON DELETE CASCADE, + idempotency_key TEXT NOT NULL, + operation TEXT NOT NULL, + request_hash TEXT NOT NULL, + actor TEXT NOT NULL, + result_id TEXT NOT NULL DEFAULT '', + created_at TIMESTAMPTZ NOT NULL, + PRIMARY KEY (work_item_id, idempotency_key) +); +-- Additive columns for databases created by an earlier revision. +ALTER TABLE work_item_create_requests ADD COLUMN IF NOT EXISTS actor TEXT NOT NULL DEFAULT ''; +ALTER TABLE work_item_requests ADD COLUMN IF NOT EXISTS actor TEXT NOT NULL DEFAULT ''; +CREATE INDEX IF NOT EXISTS work_item_create_requests_item ON work_item_create_requests (work_item_id); +` + +const itemColumns = `id, tenant, owner_principal, visibility, origin_surface, title, external_key, + state, waiting_reason, superseded_by, state_version, created_by, created_at, updated_at, + completed_at, schema_version` + +const executionColumns = `id, work_item_id, engine, engine_ref, attempt, resumed_from_execution_id, + phase, started_at, ended_at` + +// Operations recorded in work_item_requests. +const ( + opTransition = "transition" + opResume = "resume" + opIntent = "intent" + opExecution = "execution" +) + +// Store is the Postgres-backed work-item store. Every mutation of one item +// runs in one transaction under pg_advisory_xact_lock(hashtext('workitem:' || +// id)), the agentregistry.promote pattern, so two surfaces racing on the same +// item resolve to one effect and one idempotent replay or 409. +type Store struct { + pool *pgxpool.Pool + now func() time.Time +} + +// NewStore connects and creates the schema if needed. +func NewStore(ctx context.Context, connStr string) (*Store, error) { + pool, err := pgxpool.New(ctx, connStr) + if err != nil { + return nil, fmt.Errorf("workitems: connect: %w", err) + } + if _, err := pool.Exec(ctx, schema); err != nil { + pool.Close() + return nil, fmt.Errorf("workitems: create schema: %w", err) + } + slog.Info("work-items store initialized") + return &Store{pool: pool, now: func() time.Time { return time.Now().UTC() }}, nil +} + +// Close releases the pool. +func (s *Store) Close() { s.pool.Close() } + +func (s *Store) withTx(ctx context.Context, lockKey string, fn func(pgx.Tx) error) error { + tx, err := s.pool.Begin(ctx) + if err != nil { + return fmt.Errorf("workitems: begin: %w", err) + } + defer func() { _ = tx.Rollback(ctx) }() // no-op once Commit has succeeded + if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtext($1))`, lockKey); err != nil { + return fmt.Errorf("workitems: acquire lock: %w", err) + } + if err := fn(tx); err != nil { + return err + } + if err := tx.Commit(ctx); err != nil { + return fmt.Errorf("workitems: commit: %w", err) + } + return nil +} + +func scanItem(row pgx.Row) (*WorkItem, error) { + var w WorkItem + if err := row.Scan(&w.ID, &w.Tenant, &w.OwnerPrincipal, &w.Visibility, &w.OriginSurface, &w.Title, + &w.ExternalKey, &w.State, &w.WaitingReason, &w.SupersededBy, &w.StateVersion, &w.CreatedBy, + &w.CreatedAt, &w.UpdatedAt, &w.CompletedAt, &w.SchemaVersion); err != nil { + return nil, err + } + w.CreatedAt = w.CreatedAt.UTC() + w.UpdatedAt = w.UpdatedAt.UTC() + if w.CompletedAt != nil { + t := w.CompletedAt.UTC() + w.CompletedAt = &t + } + return &w, nil +} + +func scanExecution(row pgx.Row) (*Execution, error) { + var e Execution + if err := row.Scan(&e.ID, &e.WorkItemID, &e.Engine, &e.EngineRef, &e.Attempt, + &e.ResumedFromExecutionID, &e.Phase, &e.StartedAt, &e.EndedAt); err != nil { + return nil, err + } + e.StartedAt = e.StartedAt.UTC() + if e.EndedAt != nil { + t := e.EndedAt.UTC() + e.EndedAt = &t + } + return &e, nil +} + +type querier interface { + QueryRow(ctx context.Context, sql string, args ...any) pgx.Row +} + +func getItem(ctx context.Context, q querier, id string) (*WorkItem, error) { + w, err := scanItem(q.QueryRow(ctx, `SELECT `+itemColumns+` FROM work_items WHERE id=$1`, id)) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrNotFound + } + if err != nil { + return nil, fmt.Errorf("workitems: get: %w", err) + } + return w, nil +} + +// Get returns one work item or ErrNotFound. It applies no authorization: the +// caller checks tenant and visibility, as List does through Tenants and +// Viewer. +func (s *Store) Get(ctx context.Context, id string) (*WorkItem, error) { + return getItem(ctx, s.pool, id) +} + +// Create opens a work item, or returns the one this request already opened +// (same tenant and idempotency key) or the open one already carrying its +// external_key. created is false on either replay. +func (s *Store) Create(ctx context.Context, in CreateInput) (*WorkItem, bool, error) { + if err := in.validate(); err != nil { + return nil, false, err + } + var out *WorkItem + created := false + // One lock per tenant: the key lookup, the external_key lookup and the + // insert (plus recording the key) must be one decision. The partial + // unique index backs the dedupe, but ON CONFLICT alone could not also + // record the idempotency key against the item it resolved to. + err := s.withTx(ctx, "workitem-create:"+in.Tenant, func(tx pgx.Tx) error { + now := s.now() + if in.IdempotencyKey != "" { + var id, hash, actor string + err := tx.QueryRow(ctx, `SELECT work_item_id, request_hash, actor FROM work_item_create_requests + WHERE tenant=$1 AND idempotency_key=$2`, in.Tenant, in.IdempotencyKey).Scan(&id, &hash, &actor) + if err == nil { + // A key names one principal's request: another principal + // sending the same key (and body) is not a retry, and must + // not be told the first principal's write was theirs. + if hash != in.RequestHash || actor != in.Actor { + return ErrIdempotencyKeyReuse + } + w, err := getItem(ctx, tx, id) + out = w + return err + } + if !errors.Is(err, pgx.ErrNoRows) { + return fmt.Errorf("workitems: create: idempotency lookup: %w", err) + } + } + if in.ExternalKey != "" { + w, err := scanItem(tx.QueryRow(ctx, `SELECT `+itemColumns+` FROM work_items + WHERE tenant=$1 AND external_key=$2 AND state IN ('active','waiting')`, in.Tenant, in.ExternalKey)) + if err == nil { + // Dedupe only onto an item this caller could open itself: the + // same visibility, and not someone else's private item. An + // external_key is guessable (an issue URL), so anything else + // would read another principal's private work, or silently + // hand a private request a tenant-visible item. + if w.Visibility != in.Visibility || (w.Visibility == VisibilityPrivate && w.OwnerPrincipal != in.Actor) { + return ErrExternalKeyInUse + } + out = w + // The key now names this item, so a retry keeps returning it + // even after it goes terminal. + return rememberCreate(ctx, tx, in, w.ID, now) + } + if !errors.Is(err, pgx.ErrNoRows) { + return fmt.Errorf("workitems: create: external key lookup: %w", err) + } + } + + w, err := scanItem(tx.QueryRow(ctx, `INSERT INTO work_items (`+itemColumns+`) + VALUES ($1,$2,$3,$4,$5,$6,$7,'`+StateActive+`','','',1,$3,$8,$8,NULL,$9) + RETURNING `+itemColumns, + WorkItemIDPrefix+uuid.NewString(), in.Tenant, in.Actor, in.Visibility, in.OriginSurface, + in.Title, in.ExternalKey, now, SchemaVersion)) + if err != nil { + return fmt.Errorf("workitems: create: %w", err) + } + if err := appendEvent(ctx, tx, w.ID, EventCreated, "", StateActive, in.Mutation, nil, now); err != nil { + return err + } + out, created = w, true + return rememberCreate(ctx, tx, in, w.ID, now) + }) + if err != nil { + return nil, false, err + } + return out, created, nil +} + +func rememberCreate(ctx context.Context, tx pgx.Tx, in CreateInput, id string, at time.Time) error { + if in.IdempotencyKey == "" { + return nil + } + _, err := tx.Exec(ctx, `INSERT INTO work_item_create_requests + (tenant, idempotency_key, request_hash, actor, work_item_id, created_at) VALUES ($1,$2,$3,$4,$5,$6)`, + in.Tenant, in.IdempotencyKey, in.RequestHash, in.Actor, id, at) + if err != nil { + return fmt.Errorf("workitems: create: record idempotency key: %w", err) + } + return nil +} + +func appendEvent(ctx context.Context, tx pgx.Tx, id, kind, from, to string, m Mutation, detail any, at time.Time) error { + var raw []byte + if detail != nil { + b, err := json.Marshal(detail) + if err != nil { + return fmt.Errorf("workitems: event detail: %w", err) + } + raw = b + } + _, err := tx.Exec(ctx, `INSERT INTO work_item_events + (work_item_id, seq, kind, from_state, to_state, actor_principal, surface, request_id, detail, created_at) + VALUES ($1, (SELECT COALESCE(MAX(seq), 0) + 1 FROM work_item_events WHERE work_item_id=$1), + $2,$3,$4,$5,$6,$7,$8,$9)`, + id, kind, from, to, m.Actor, m.Surface, m.RequestID, raw, at) + if err != nil { + return fmt.Errorf("workitems: append event: %w", err) + } + return nil +} + +// replayed looks up a per-item idempotency key. It returns the recorded +// result id and true on a replay of the same request, ErrIdempotencyKeyReuse +// when the key was used for something else, and false when the key is new. +func replayed(ctx context.Context, tx pgx.Tx, id, op string, m Mutation) (string, bool, error) { + if m.IdempotencyKey == "" { + return "", false, nil + } + var gotOp, hash, actor, result string + err := tx.QueryRow(ctx, `SELECT operation, request_hash, actor, result_id FROM work_item_requests + WHERE work_item_id=$1 AND idempotency_key=$2`, id, m.IdempotencyKey).Scan(&gotOp, &hash, &actor, &result) + if errors.Is(err, pgx.ErrNoRows) { + return "", false, nil + } + if err != nil { + return "", false, fmt.Errorf("workitems: idempotency lookup: %w", err) + } + if gotOp != op || hash != m.RequestHash || actor != m.Actor { + return "", false, ErrIdempotencyKeyReuse + } + return result, true, nil +} + +func remember(ctx context.Context, tx pgx.Tx, id, op, result string, m Mutation, at time.Time) error { + if m.IdempotencyKey == "" { + return nil + } + _, err := tx.Exec(ctx, `INSERT INTO work_item_requests + (work_item_id, idempotency_key, operation, request_hash, actor, result_id, created_at) + VALUES ($1,$2,$3,$4,$5,$6,$7)`, id, m.IdempotencyKey, op, m.RequestHash, m.Actor, result, at) + if err != nil { + return fmt.Errorf("workitems: record idempotency key: %w", err) + } + return nil +} + +// Transition applies one lifecycle action. A replay of the same request +// returns the item as it is now. +func (s *Store) Transition(ctx context.Context, in TransitionInput) (*WorkItem, error) { + if err := in.validate(); err != nil { + return nil, err + } + var out *WorkItem + err := s.withTx(ctx, "workitem:"+in.WorkItemID, func(tx pgx.Tx) error { + cur, err := getItem(ctx, tx, in.WorkItemID) + if err != nil { + return err + } + if _, ok, err := replayed(ctx, tx, cur.ID, opTransition, in.Mutation); err != nil || ok { + out = cur + return err + } + if cur.StateVersion != in.ExpectedStateVersion { + return &ConflictError{Err: ErrVersionConflict, Current: cur} + } + to, err := Next(cur.State, in.Action) + if err != nil { + return &ConflictError{Err: err, Current: cur} + } + if in.Action == ActionSupersede { + next, err := getItem(ctx, tx, in.SupersededBy) + if errors.Is(err, ErrNotFound) || (err == nil && next.Tenant != cur.Tenant) { + return invalid("superseded_by %s is not a work item in tenant %s", in.SupersededBy, cur.Tenant) + } + if err != nil { + return err + } + } + now := s.now() + var completedAt *time.Time + if IsTerminal(to) { + completedAt = &now + } + // The version guard repeats the check above in SQL: it holds even + // if a caller ever reaches this without the advisory lock. + w, err := scanItem(tx.QueryRow(ctx, `UPDATE work_items SET state=$2, waiting_reason=$3, + superseded_by=$4, state_version=state_version+1, updated_at=$5, completed_at=$6 + WHERE id=$1 AND state_version=$7 RETURNING `+itemColumns, + cur.ID, to, in.WaitingReason, in.SupersededBy, now, completedAt, cur.StateVersion)) + if errors.Is(err, pgx.ErrNoRows) { + return &ConflictError{Err: ErrVersionConflict, Current: cur} + } + if err != nil { + return fmt.Errorf("workitems: transition: %w", err) + } + detail := map[string]string{"action": in.Action} + if in.WaitingReason != "" { + detail["waiting_reason"] = in.WaitingReason + } + if in.SupersededBy != "" { + detail["superseded_by"] = in.SupersededBy + } + if err := appendEvent(ctx, tx, cur.ID, EventStateChanged, cur.State, to, in.Mutation, detail, now); err != nil { + return err + } + if err := remember(ctx, tx, cur.ID, opTransition, "", in.Mutation, now); err != nil { + return err + } + out = w + return nil + }) + if err != nil { + return nil, err + } + return out, nil +} + +// Resume starts a new execution continuing a prior one. It refuses while an +// execution is still Pending or Running, and on a terminal item. A replay +// returns the execution the first request started. +func (s *Store) Resume(ctx context.Context, in ResumeInput) (*WorkItem, *Execution, bool, error) { + if err := in.validate(); err != nil { + return nil, nil, false, err + } + var item *WorkItem + var exec *Execution + created := false + err := s.withTx(ctx, "workitem:"+in.WorkItemID, func(tx pgx.Tx) error { + cur, err := getItem(ctx, tx, in.WorkItemID) + if err != nil { + return err + } + if result, ok, err := replayed(ctx, tx, cur.ID, opResume, in.Mutation); err != nil { + return err + } else if ok { + item = cur + exec, err = getExecution(ctx, tx, cur.ID, result) + return err + } + if cur.StateVersion != in.ExpectedStateVersion { + return &ConflictError{Err: ErrVersionConflict, Current: cur} + } + if IsTerminal(cur.State) { + return &ConflictError{Err: fmt.Errorf("%w: resume from %s", ErrInvalidTransition, cur.State), Current: cur} + } + execs, err := listExecutions(ctx, tx, cur.ID) + if err != nil { + return err + } + from := in.ResumedFromExecutionID + for i := range execs { + if !IsTerminalPhase(execs[i].Phase) { + return &ConflictError{Err: ErrExecutionActive, Current: cur} + } + } + if from == "" && len(execs) > 0 { + from = execs[len(execs)-1].ID + } + // An active item starts its first run with AttachExecution; only a + // waiting one may resume with nothing before it, since Resume is + // the sole way out of waiting. + if len(execs) == 0 && cur.State == StateActive { + return invalid("%s has no execution to resume; attach the first one", cur.ID) + } + if from != "" && !containsExecution(execs, from) { + return invalid("resumed_from_execution_id %s is not an execution of %s", from, cur.ID) + } + for i := range execs { + if execs[i].Engine == in.Engine && execs[i].EngineRef == in.EngineRef { + return invalid("%s %s is already execution %s of %s; a resume starts a new run", + in.Engine, in.EngineRef, execs[i].ID, cur.ID) + } + } + now := s.now() + exec, err = insertExecution(ctx, tx, cur.ID, in.Engine, in.EngineRef, PhasePending, from, len(execs)+1, now) + if err != nil { + return err + } + to := cur.State + if cur.State == StateWaiting { + if to, err = Next(cur.State, ActionResume); err != nil { + return &ConflictError{Err: err, Current: cur} + } + } + item, err = scanItem(tx.QueryRow(ctx, `UPDATE work_items SET state=$2, waiting_reason='', + state_version=state_version+1, updated_at=$3 + WHERE id=$1 AND state_version=$4 RETURNING `+itemColumns, cur.ID, to, now, cur.StateVersion)) + if errors.Is(err, pgx.ErrNoRows) { + return &ConflictError{Err: ErrVersionConflict, Current: cur} + } + if err != nil { + return fmt.Errorf("workitems: resume: %w", err) + } + detail := map[string]string{"execution_id": exec.ID} + if from != "" { + detail["resumed_from_execution_id"] = from + } + if err := appendEvent(ctx, tx, cur.ID, EventResumed, cur.State, to, in.Mutation, detail, now); err != nil { + return err + } + if err := remember(ctx, tx, cur.ID, opResume, exec.ID, in.Mutation, now); err != nil { + return err + } + created = true + return nil + }) + if err != nil { + return nil, nil, false, err + } + return item, exec, created, nil +} + +func containsExecution(execs []Execution, id string) bool { + for i := range execs { + if execs[i].ID == id { + return true + } + } + return false +} + +func insertExecution(ctx context.Context, tx pgx.Tx, itemID, engine, ref, phase, from string, attempt int, now time.Time) (*Execution, error) { + var ended *time.Time + if IsTerminalPhase(phase) { + ended = &now + } + e, err := scanExecution(tx.QueryRow(ctx, `INSERT INTO work_item_executions (`+executionColumns+`) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9) RETURNING `+executionColumns, + ExecutionIDPrefix+uuid.NewString(), itemID, engine, ref, attempt, from, phase, now, ended)) + if err != nil { + return nil, fmt.Errorf("workitems: insert execution: %w", err) + } + return e, nil +} + +func getExecution(ctx context.Context, q querier, itemID, id string) (*Execution, error) { + e, err := scanExecution(q.QueryRow(ctx, `SELECT `+executionColumns+` FROM work_item_executions + WHERE work_item_id=$1 AND id=$2`, itemID, id)) + if errors.Is(err, pgx.ErrNoRows) { + return nil, fmt.Errorf("workitems: execution %s of %s: %w", id, itemID, ErrNotFound) + } + if err != nil { + return nil, fmt.Errorf("workitems: get execution: %w", err) + } + return e, nil +} + +type rowsQuerier interface { + Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error) +} + +func listExecutions(ctx context.Context, q rowsQuerier, itemID string) ([]Execution, error) { + rows, err := q.Query(ctx, `SELECT `+executionColumns+` FROM work_item_executions + WHERE work_item_id=$1 ORDER BY attempt`, itemID) + if err != nil { + return nil, fmt.Errorf("workitems: list executions: %w", err) + } + defer rows.Close() + out := []Execution{} + for rows.Next() { + e, err := scanExecution(rows) + if err != nil { + return nil, fmt.Errorf("workitems: list executions: %w", err) + } + out = append(out, *e) + } + return out, rows.Err() +} + +// Executions lists a work item's executions, oldest attempt first. No +// authorization: the caller checks tenant and visibility. +func (s *Store) Executions(ctx context.Context, itemID string) ([]Execution, error) { + return listExecutions(ctx, s.pool, itemID) +} + +// AttachExecution correlates an engine run with a non-terminal work item, +// or records a later phase of one already attached under the same +// (engine, engine_ref). A finished execution's phase never changes. +func (s *Store) AttachExecution(ctx context.Context, in ExecutionInput) (*Execution, bool, error) { + if err := in.validate(); err != nil { + return nil, false, err + } + var out *Execution + created := false + err := s.withTx(ctx, "workitem:"+in.WorkItemID, func(tx pgx.Tx) error { + cur, err := getItem(ctx, tx, in.WorkItemID) + if err != nil { + return err + } + if result, ok, err := replayed(ctx, tx, cur.ID, opExecution, in.Mutation); err != nil { + return err + } else if ok { + out, err = getExecution(ctx, tx, cur.ID, result) + return err + } + now := s.now() + existing, err := scanExecution(tx.QueryRow(ctx, `SELECT `+executionColumns+` FROM work_item_executions + WHERE work_item_id=$1 AND engine=$2 AND engine_ref=$3`, cur.ID, in.Engine, in.EngineRef)) + switch { + case err == nil: + if existing.Phase != in.Phase { + if IsTerminalPhase(existing.Phase) { + return &ConflictError{ + Err: fmt.Errorf("%w: execution %s already ended %s", ErrInvalidTransition, existing.ID, existing.Phase), + Current: cur, + } + } + var ended *time.Time + if IsTerminalPhase(in.Phase) { + ended = &now + } + existing, err = scanExecution(tx.QueryRow(ctx, `UPDATE work_item_executions SET phase=$2, ended_at=$3 + WHERE id=$1 RETURNING `+executionColumns, existing.ID, in.Phase, ended)) + if err != nil { + return fmt.Errorf("workitems: update execution: %w", err) + } + } + out = existing + case errors.Is(err, pgx.ErrNoRows): + if IsTerminal(cur.State) { + return &ConflictError{Err: fmt.Errorf("%w: attach to %s", ErrInvalidTransition, cur.State), Current: cur} + } + execs, err := listExecutions(ctx, tx, cur.ID) + if err != nil { + return err + } + if !IsTerminalPhase(in.Phase) { + for i := range execs { + if !IsTerminalPhase(execs[i].Phase) { + return &ConflictError{Err: ErrExecutionActive, Current: cur} + } + } + } + out, err = insertExecution(ctx, tx, cur.ID, in.Engine, in.EngineRef, in.Phase, "", len(execs)+1, now) + if err != nil { + return err + } + detail := map[string]string{"execution_id": out.ID, "engine": in.Engine, "engine_ref": in.EngineRef} + if err := appendEvent(ctx, tx, cur.ID, EventExecutionAttached, "", "", in.Mutation, detail, now); err != nil { + return err + } + created = true + default: + return fmt.Errorf("workitems: find execution: %w", err) + } + return remember(ctx, tx, cur.ID, opExecution, out.ID, in.Mutation, now) + }) + if err != nil { + return nil, false, err + } + return out, created, nil +} + +// AppendIntent records a bounded intent on a non-terminal work item. +func (s *Store) AppendIntent(ctx context.Context, in IntentInput) (*Intent, bool, error) { + if err := in.validate(); err != nil { + return nil, false, err + } + var out *Intent + created := false + err := s.withTx(ctx, "workitem:"+in.WorkItemID, func(tx pgx.Tx) error { + cur, err := getItem(ctx, tx, in.WorkItemID) + if err != nil { + return err + } + if result, ok, err := replayed(ctx, tx, cur.ID, opIntent, in.Mutation); err != nil { + return err + } else if ok { + out, err = getIntent(ctx, tx, cur.ID, result) + return err + } + if IsTerminal(cur.State) { + return &ConflictError{Err: fmt.Errorf("%w: intent on %s", ErrInvalidTransition, cur.State), Current: cur} + } + now := s.now() + var params []byte + if len(in.Params) > 0 { + params = in.Params + } + var intent Intent + var text *string + err = tx.QueryRow(ctx, `INSERT INTO work_item_intents + (work_item_id, actor_principal, surface, text, params, created_at) + VALUES ($1,$2,$3,$4,$5,$6) + RETURNING id, work_item_id, actor_principal, surface, text, params, created_at`, + cur.ID, in.Actor, in.Surface, in.Text, params, now).Scan( + &intent.ID, &intent.WorkItemID, &intent.ActorPrincipal, &intent.Surface, &text, &intent.Params, &intent.CreatedAt) + if err != nil { + return fmt.Errorf("workitems: append intent: %w", err) + } + if text != nil { + intent.Text = *text + } + intent.CreatedAt = intent.CreatedAt.UTC() + // The event names the intent; it never copies the text. + if err := appendEvent(ctx, tx, cur.ID, EventIntentAppended, "", "", in.Mutation, + map[string]int64{"intent_id": intent.ID}, now); err != nil { + return err + } + if err := remember(ctx, tx, cur.ID, opIntent, fmt.Sprint(intent.ID), in.Mutation, now); err != nil { + return err + } + out, created = &intent, true + return nil + }) + if err != nil { + return nil, false, err + } + return out, created, nil +} + +func getIntent(ctx context.Context, q querier, itemID, id string) (*Intent, error) { + var intent Intent + var text *string + err := q.QueryRow(ctx, `SELECT id, work_item_id, actor_principal, surface, text, params, created_at + FROM work_item_intents WHERE work_item_id=$1 AND id::text=$2`, itemID, id).Scan( + &intent.ID, &intent.WorkItemID, &intent.ActorPrincipal, &intent.Surface, &text, &intent.Params, &intent.CreatedAt) + if errors.Is(err, pgx.ErrNoRows) { + return nil, fmt.Errorf("workitems: intent %s of %s: %w", id, itemID, ErrNotFound) + } + if err != nil { + return nil, fmt.Errorf("workitems: get intent: %w", err) + } + if text != nil { + intent.Text = *text + } + intent.CreatedAt = intent.CreatedAt.UTC() + return &intent, nil +} + +// LinkSurface correlates a surface-native conversation with a work item, or +// refreshes last_seen_at on an existing correlation. The upsert on +// (item, surface, external_id) is its idempotency: an Idempotency-Key adds +// nothing here and is not recorded. +func (s *Store) LinkSurface(ctx context.Context, in SurfaceRefInput) (*SurfaceRef, bool, error) { + if err := in.validate(); err != nil { + return nil, false, err + } + var out SurfaceRef + created := false + err := s.withTx(ctx, "workitem:"+in.WorkItemID, func(tx pgx.Tx) error { + cur, err := getItem(ctx, tx, in.WorkItemID) + if err != nil { + return err + } + now := s.now() + var inserted bool + err = tx.QueryRow(ctx, `INSERT INTO work_item_surface_refs + (work_item_id, surface, external_id, actor_external_id, first_seen_at, last_seen_at) + VALUES ($1,$2,$3,$4,$5,$5) + ON CONFLICT (work_item_id, surface, external_id) DO UPDATE SET + last_seen_at=EXCLUDED.last_seen_at, + actor_external_id=CASE WHEN EXCLUDED.actor_external_id <> '' THEN EXCLUDED.actor_external_id + ELSE work_item_surface_refs.actor_external_id END + RETURNING work_item_id, surface, external_id, actor_external_id, first_seen_at, last_seen_at, (xmax = 0)`, + cur.ID, in.Surface, in.ExternalID, in.ActorExternalID, now).Scan( + &out.WorkItemID, &out.Surface, &out.ExternalID, &out.ActorExternalID, &out.FirstSeenAt, &out.LastSeenAt, &inserted) + if err != nil { + return fmt.Errorf("workitems: link surface: %w", err) + } + out.FirstSeenAt, out.LastSeenAt = out.FirstSeenAt.UTC(), out.LastSeenAt.UTC() + if inserted { + // Correlation metadata only: the event names the surface, never + // the external ids (contract "Retention and privacy"). + if err := appendEvent(ctx, tx, cur.ID, EventSurfaceLinked, "", "", in.Mutation, nil, now); err != nil { + return err + } + created = true + } + return nil + }) + if err != nil { + return nil, false, err + } + return &out, created, nil +} + +// Events returns a work item's lifecycle history in order. No +// authorization: the caller checks tenant and visibility. +func (s *Store) Events(ctx context.Context, itemID string) ([]Event, error) { + rows, err := s.pool.Query(ctx, `SELECT work_item_id, seq, kind, from_state, to_state, actor_principal, + surface, request_id, detail, created_at + FROM work_item_events WHERE work_item_id=$1 ORDER BY seq`, itemID) + if err != nil { + return nil, fmt.Errorf("workitems: events: %w", err) + } + defer rows.Close() + out := []Event{} + for rows.Next() { + var e Event + if err := rows.Scan(&e.WorkItemID, &e.Seq, &e.Kind, &e.FromState, &e.ToState, &e.ActorPrincipal, + &e.Surface, &e.RequestID, &e.Detail, &e.CreatedAt); err != nil { + return nil, fmt.Errorf("workitems: events: %w", err) + } + e.CreatedAt = e.CreatedAt.UTC() + out = append(out, e) + } + return out, rows.Err() +} + +// List returns work items matching f, most recently updated first. +func (s *Store) List(ctx context.Context, f ListFilter) ([]WorkItem, error) { + state := f.State + if state == "" { + state = FilterOpen + } + if state != FilterOpen && !validState(state) { + return nil, invalid("unknown state filter %q", state) + } + limit := f.Limit + if limit <= 0 { + limit = defaultListLimit + } + if limit > maxListLimit { + limit = maxListLimit + } + query := `SELECT ` + itemColumns + ` FROM work_items WHERE TRUE` + args := []any{} + arg := func(v any) string { + args = append(args, v) + return fmt.Sprintf("$%d", len(args)) + } + // Fail closed: a filter that names no tenant and does not ask for all + // of them matches nothing. AllTenants only lifts that requirement; named + // Tenants still narrow. + if len(f.Tenants) > 0 { + query += ` AND tenant = ANY(` + arg(f.Tenants) + `)` + } else if !f.AllTenants { + return []WorkItem{}, nil + } + if state == FilterOpen { + query += ` AND state IN ('active','waiting')` + } else { + query += ` AND state = ` + arg(state) + } + if f.Owner != "" { + query += ` AND owner_principal = ` + arg(f.Owner) + } + if f.Viewer != "" { + query += ` AND (visibility = 'tenant' OR owner_principal = ` + arg(f.Viewer) + `)` + } + query += ` ORDER BY updated_at DESC, id LIMIT ` + arg(limit) + rows, err := s.pool.Query(ctx, query, args...) + if err != nil { + return nil, fmt.Errorf("workitems: list: %w", err) + } + defer rows.Close() + out := []WorkItem{} + for rows.Next() { + w, err := scanItem(rows) + if err != nil { + return nil, fmt.Errorf("workitems: list: %w", err) + } + out = append(out, *w) + } + return out, rows.Err() +} diff --git a/internal/workitems/store_test.go b/internal/workitems/store_test.go new file mode 100644 index 00000000..4efae24a --- /dev/null +++ b/internal/workitems/store_test.go @@ -0,0 +1,737 @@ +package workitems + +import ( + "context" + "errors" + "os" + "strings" + "sync" + "testing" +) + +func newStoreForTest(t *testing.T) *Store { + t.Helper() + connStr := os.Getenv("TEST_DATABASE_URL") + if connStr == "" { + t.Skip("TEST_DATABASE_URL not set; skipping Postgres-backed work-items store test") + } + ctx := context.Background() + s, err := NewStore(ctx, connStr) + if err != nil { + t.Fatalf("NewStore: %v", err) + } + wipe := func() { + // Every other table cascades from work_items. + if _, err := s.pool.Exec(ctx, "DELETE FROM work_items"); err != nil { + t.Fatal(err) + } + } + wipe() + t.Cleanup(func() { + wipe() + s.Close() + }) + return s +} + +func as(actor string) Mutation { return Mutation{Actor: actor, Surface: "cli", RequestID: "req-1"} } + +func keyed(actor, key, hash string) Mutation { + m := as(actor) + m.IdempotencyKey, m.RequestHash = key, hash + return m +} + +func open(t *testing.T, s *Store, in CreateInput) *WorkItem { + t.Helper() + if in.Actor == "" { + in.Mutation = as("github:alice") + } + if in.Tenant == "" { + in.Tenant = "acme" + } + if in.Visibility == "" { + in.Visibility = VisibilityTenant + } + if in.OriginSurface == "" { + in.OriginSurface = "cli" + } + if in.Title == "" { + in.Title = "ship the thing" + } + w, created, err := s.Create(context.Background(), in) + if err != nil { + t.Fatalf("Create: %v", err) + } + if !created { + t.Fatalf("Create returned an existing item %s", w.ID) + } + return w +} + +func conflictCurrent(t *testing.T, err error, want error) *WorkItem { + t.Helper() + if !errors.Is(err, want) { + t.Fatalf("err = %v, want %v", err, want) + } + var ce *ConflictError + if !errors.As(err, &ce) || ce.Current == nil { + t.Fatalf("err %v does not carry the current item", err) + } + return ce.Current +} + +func TestTransitionTable(t *testing.T) { + for _, terminal := range []string{StateCompleted, StateSuperseded, StateArchived} { + if !IsTerminal(terminal) { + t.Errorf("%s is not terminal", terminal) + } + for _, action := range []string{ActionWait, ActionResume, ActionComplete, ActionSupersede, ActionArchive} { + if _, err := Next(terminal, action); !errors.Is(err, ErrInvalidTransition) { + t.Errorf("%s from %s allowed", action, terminal) + } + } + } + if _, err := Next(StateActive, ActionResume); !errors.Is(err, ErrInvalidTransition) { + t.Error("resume from active is not in the table") + } + if to, err := Next(StateWaiting, ActionResume); err != nil || to != StateActive { + t.Errorf("resume from waiting = %q, %v", to, err) + } +} + +func TestCreateAndGet(t *testing.T) { + s := newStoreForTest(t) + ctx := context.Background() + w := open(t, s, CreateInput{}) + if !strings.HasPrefix(w.ID, WorkItemIDPrefix) || w.State != StateActive || w.StateVersion != 1 || + w.SchemaVersion != SchemaVersion || w.OwnerPrincipal != "github:alice" || w.CreatedBy != "github:alice" { + t.Fatalf("created %+v", w) + } + got, err := s.Get(ctx, w.ID) + if err != nil || *got != *w { + t.Fatalf("Get = %+v, %v; want %+v", got, err, w) + } + if _, err := s.Get(ctx, "wi_missing"); !errors.Is(err, ErrNotFound) { + t.Fatalf("Get(missing) err = %v", err) + } + events, err := s.Events(ctx, w.ID) + if err != nil || len(events) != 1 || events[0].Kind != EventCreated || events[0].ToState != StateActive || + events[0].ActorPrincipal != "github:alice" || events[0].RequestID != "req-1" { + t.Fatalf("events = %+v, %v", events, err) + } +} + +func TestCreateRequiresAnActor(t *testing.T) { + s := newStoreForTest(t) + _, _, err := s.Create(context.Background(), CreateInput{Tenant: "acme", Visibility: VisibilityTenant, OriginSurface: "cli", Title: "x"}) + if !errors.Is(err, ErrInvalid) { + t.Fatalf("err = %v, want ErrInvalid", err) + } +} + +func TestCreateIsIdempotentPerTenantKey(t *testing.T) { + s := newStoreForTest(t) + ctx := context.Background() + in := CreateInput{Mutation: keyed("github:alice", "k1", "h1"), Tenant: "acme", Visibility: VisibilityTenant, OriginSurface: "telegram", Title: "a"} + first, created, err := s.Create(ctx, in) + if err != nil || !created { + t.Fatalf("first create: %v %v", created, err) + } + again, created, err := s.Create(ctx, in) + if err != nil || created || again.ID != first.ID { + t.Fatalf("replay = %v %v %v; want %s", again, created, err, first.ID) + } + in.RequestHash = "h2" + if _, _, err := s.Create(ctx, in); !errors.Is(err, ErrIdempotencyKeyReuse) { + t.Fatalf("reused key with another request: err = %v", err) + } + // The key is per tenant. + in.Tenant, in.RequestHash = "other", "h1" + if other, created, err := s.Create(ctx, in); err != nil || !created || other.ID == first.ID { + t.Fatalf("other tenant = %v %v %v", other, created, err) + } +} + +func TestExternalKeyDedupesOnlyOpenWork(t *testing.T) { + s := newStoreForTest(t) + ctx := context.Background() + w := open(t, s, CreateInput{ExternalKey: "https://github.com/mctlhq/mctl-api/issues/349"}) + dup, created, err := s.Create(ctx, CreateInput{Mutation: as("github:bob"), Tenant: "acme", Visibility: VisibilityTenant, + OriginSurface: "telegram", Title: "same issue", ExternalKey: w.ExternalKey}) + if err != nil || created || dup.ID != w.ID { + t.Fatalf("dedupe = %v %v %v", dup, created, err) + } + if _, err := s.Transition(ctx, TransitionInput{Mutation: as("github:alice"), WorkItemID: w.ID, Action: ActionComplete, ExpectedStateVersion: 1}); err != nil { + t.Fatal(err) + } + fresh := open(t, s, CreateInput{ExternalKey: w.ExternalKey}) + if fresh.ID == w.ID { + t.Fatal("a completed item still claimed its external key") + } +} + +func TestTransitionsAndVersionConflicts(t *testing.T) { + s := newStoreForTest(t) + ctx := context.Background() + w := open(t, s, CreateInput{}) + + waiting, err := s.Transition(ctx, TransitionInput{Mutation: as("github:alice"), WorkItemID: w.ID, + Action: ActionWait, WaitingReason: WaitingInput, ExpectedStateVersion: 1}) + if err != nil || waiting.State != StateWaiting || waiting.WaitingReason != WaitingInput || waiting.StateVersion != 2 { + t.Fatalf("wait = %+v, %v", waiting, err) + } + + // A stale version is a 409 carrying the current state; nothing changes. + _, err = s.Transition(ctx, TransitionInput{Mutation: as("github:alice"), WorkItemID: w.ID, Action: ActionComplete, ExpectedStateVersion: 1}) + if cur := conflictCurrent(t, err, ErrVersionConflict); cur.StateVersion != 2 || cur.State != StateWaiting { + t.Fatalf("conflict current = %+v", cur) + } + + done, err := s.Transition(ctx, TransitionInput{Mutation: as("github:alice"), WorkItemID: w.ID, Action: ActionComplete, ExpectedStateVersion: 2}) + if err != nil || done.State != StateCompleted || done.WaitingReason != "" || done.CompletedAt == nil || done.StateVersion != 3 { + t.Fatalf("complete = %+v, %v", done, err) + } + _, err = s.Transition(ctx, TransitionInput{Mutation: as("github:alice"), WorkItemID: w.ID, Action: ActionArchive, ExpectedStateVersion: 3}) + if cur := conflictCurrent(t, err, ErrInvalidTransition); cur.State != StateCompleted || cur.StateVersion != 3 { + t.Fatalf("out of terminal: current = %+v", cur) + } + if _, err := s.Transition(ctx, TransitionInput{Mutation: as("github:alice"), WorkItemID: "wi_missing", Action: ActionArchive, ExpectedStateVersion: 1}); !errors.Is(err, ErrNotFound) { + t.Fatalf("missing item: err = %v", err) + } + for _, bad := range []TransitionInput{ + {WorkItemID: w.ID, Action: ActionWait, ExpectedStateVersion: 3}, // no reason + {WorkItemID: w.ID, Action: ActionArchive, WaitingReason: WaitingInput, ExpectedStateVersion: 3}, // stray reason + {WorkItemID: w.ID, Action: ActionSupersede, ExpectedStateVersion: 3}, // no successor + {WorkItemID: w.ID, Action: ActionArchive}, // no version + {WorkItemID: w.ID, Action: ActionResume, ExpectedStateVersion: 3}, // own route + } { + bad.Mutation = as("github:alice") + if _, err := s.Transition(ctx, bad); !errors.Is(err, ErrInvalid) { + t.Errorf("%+v: err = %v, want ErrInvalid", bad, err) + } + } +} + +func TestSupersedeNeedsASuccessorInTheSameTenant(t *testing.T) { + s := newStoreForTest(t) + ctx := context.Background() + w := open(t, s, CreateInput{}) + elsewhere := open(t, s, CreateInput{Tenant: "other"}) + in := TransitionInput{Mutation: as("github:alice"), WorkItemID: w.ID, Action: ActionSupersede, SupersededBy: elsewhere.ID, ExpectedStateVersion: 1} + if _, err := s.Transition(ctx, in); !errors.Is(err, ErrInvalid) { + t.Fatalf("cross-tenant successor: err = %v", err) + } + next := open(t, s, CreateInput{}) + in.SupersededBy = next.ID + got, err := s.Transition(ctx, in) + if err != nil || got.State != StateSuperseded || got.SupersededBy != next.ID { + t.Fatalf("supersede = %+v, %v", got, err) + } +} + +func TestTransitionReplayAppliesOnce(t *testing.T) { + s := newStoreForTest(t) + ctx := context.Background() + w := open(t, s, CreateInput{}) + in := TransitionInput{Mutation: keyed("github:alice", "t1", "h"), WorkItemID: w.ID, Action: ActionWait, WaitingReason: WaitingApproval, ExpectedStateVersion: 1} + first, err := s.Transition(ctx, in) + if err != nil { + t.Fatal(err) + } + // The retry still names version 1; it is a replay, not a conflict. + again, err := s.Transition(ctx, in) + if err != nil || again.StateVersion != first.StateVersion { + t.Fatalf("replay = %+v, %v", again, err) + } + in.RequestHash = "different" + if _, err := s.Transition(ctx, in); !errors.Is(err, ErrIdempotencyKeyReuse) { + t.Fatalf("reused key: err = %v", err) + } + events, _ := s.Events(ctx, w.ID) + if len(events) != 2 { + t.Fatalf("events = %+v; the replay must not append", events) + } +} + +func TestConcurrentTransitionsResolveToExactlyOne(t *testing.T) { + s := newStoreForTest(t) + ctx := context.Background() + const racers, rounds = 16, 5 + for range rounds { + w := open(t, s, CreateInput{}) + var ready, done sync.WaitGroup + start := make(chan struct{}) + errs := make([]error, racers) + for i := range racers { + ready.Add(1) + done.Add(1) + go func() { + defer done.Done() + ready.Done() + <-start + _, errs[i] = s.Transition(ctx, TransitionInput{Mutation: as("github:alice"), WorkItemID: w.ID, + Action: ActionWait, WaitingReason: WaitingInput, ExpectedStateVersion: 1}) + }() + } + ready.Wait() + close(start) + done.Wait() + won := 0 + for _, err := range errs { + switch { + case err == nil: + won++ + case errors.Is(err, ErrVersionConflict): + default: + t.Fatalf("unexpected error: %v", err) + } + } + if won != 1 { + t.Fatalf("%d transitions won, want exactly 1", won) + } + got, _ := s.Get(ctx, w.ID) + events, _ := s.Events(ctx, w.ID) + if got.StateVersion != 2 || len(events) != 2 { + t.Fatalf("state_version = %d with %d events, want 2 and 2", got.StateVersion, len(events)) + } + } +} + +func TestExecutionsAttachCorrelateAndStayUnique(t *testing.T) { + s := newStoreForTest(t) + ctx := context.Background() + w := open(t, s, CreateInput{}) + agent := as("service:mctl-agent") + + run, created, err := s.AttachExecution(ctx, ExecutionInput{Mutation: agent, WorkItemID: w.ID, Engine: EngineTemporal, EngineRef: "dev-loop-1", Phase: PhaseRunning}) + if err != nil || !created || run.Attempt != 1 || !strings.HasPrefix(run.ID, ExecutionIDPrefix) || run.EndedAt != nil { + t.Fatalf("attach = %+v %v %v", run, created, err) + } + // A second live execution is refused. + _, _, err = s.AttachExecution(ctx, ExecutionInput{Mutation: agent, WorkItemID: w.ID, Engine: EngineArgo, EngineRef: "wf-2", Phase: PhasePending}) + conflictCurrent(t, err, ErrExecutionActive) + + // The same (engine, ref) correlates a later phase instead of duplicating. + ended, created, err := s.AttachExecution(ctx, ExecutionInput{Mutation: agent, WorkItemID: w.ID, Engine: EngineTemporal, EngineRef: "dev-loop-1", Phase: PhaseSucceeded}) + if err != nil || created || ended.ID != run.ID || ended.Phase != PhaseSucceeded || ended.EndedAt == nil { + t.Fatalf("correlate = %+v %v %v", ended, created, err) + } + _, _, err = s.AttachExecution(ctx, ExecutionInput{Mutation: agent, WorkItemID: w.ID, Engine: EngineTemporal, EngineRef: "dev-loop-1", Phase: PhaseRunning}) + conflictCurrent(t, err, ErrInvalidTransition) + + if _, _, err := s.AttachExecution(ctx, ExecutionInput{Mutation: agent, WorkItemID: w.ID, Engine: "k8s", EngineRef: "x", Phase: PhaseRunning}); !errors.Is(err, ErrInvalid) { + t.Fatalf("unknown engine: err = %v", err) + } + execs, err := s.Executions(ctx, w.ID) + if err != nil || len(execs) != 1 { + t.Fatalf("executions = %+v, %v", execs, err) + } +} + +func TestResumeContinuesAPriorExecution(t *testing.T) { + s := newStoreForTest(t) + ctx := context.Background() + w := open(t, s, CreateInput{}) + agent := as("service:mctl-agent") + first, _, err := s.AttachExecution(ctx, ExecutionInput{Mutation: agent, WorkItemID: w.ID, Engine: EngineTemporal, EngineRef: "run-1", Phase: PhaseRunning}) + if err != nil { + t.Fatal(err) + } + waiting, err := s.Transition(ctx, TransitionInput{Mutation: agent, WorkItemID: w.ID, Action: ActionWait, WaitingReason: WaitingInput, ExpectedStateVersion: 1}) + if err != nil { + t.Fatal(err) + } + resume := ResumeInput{Mutation: keyed("github:alice", "r1", "h"), WorkItemID: w.ID, ExpectedStateVersion: waiting.StateVersion, Engine: EngineTemporal, EngineRef: "run-2"} + + // Not while the prior execution is still running. + _, _, _, err = s.Resume(ctx, resume) + conflictCurrent(t, err, ErrExecutionActive) + + if _, _, err := s.AttachExecution(ctx, ExecutionInput{Mutation: agent, WorkItemID: w.ID, Engine: EngineTemporal, EngineRef: "run-1", Phase: PhaseFailed}); err != nil { + t.Fatal(err) + } + item, exec, created, err := s.Resume(ctx, resume) + if err != nil || !created { + t.Fatalf("resume: %v %v", created, err) + } + if item.State != StateActive || item.WaitingReason != "" || item.StateVersion != waiting.StateVersion+1 { + t.Fatalf("resumed item = %+v", item) + } + if exec.Attempt != 2 || exec.ResumedFromExecutionID != first.ID || exec.Phase != PhasePending || exec.EngineRef != "run-2" { + t.Fatalf("resumed execution = %+v", exec) + } + + // A retry of the same resume returns the same execution, even though + // its expected version is now stale. + _, again, created, err := s.Resume(ctx, resume) + if err != nil || created || again.ID != exec.ID { + t.Fatalf("replay = %+v %v %v", again, created, err) + } + events, _ := s.Events(ctx, w.ID) + last := events[len(events)-1] + if last.Kind != EventResumed || last.FromState != StateWaiting || last.ToState != StateActive || last.ActorPrincipal != "github:alice" { + t.Fatalf("last event = %+v", last) + } + for _, e := range events { + if e.Kind == "resumed" && e.ToState == "resumed" { + t.Fatal("resumed must never be a state") + } + } + + // A terminal item cannot be resumed. + if _, _, err := s.AttachExecution(ctx, ExecutionInput{Mutation: agent, WorkItemID: w.ID, Engine: EngineTemporal, EngineRef: "run-2", Phase: PhaseSucceeded}); err != nil { + t.Fatal(err) + } + done, err := s.Transition(ctx, TransitionInput{Mutation: agent, WorkItemID: w.ID, Action: ActionComplete, ExpectedStateVersion: item.StateVersion}) + if err != nil { + t.Fatal(err) + } + _, _, _, err = s.Resume(ctx, ResumeInput{Mutation: as("github:alice"), WorkItemID: w.ID, ExpectedStateVersion: done.StateVersion, Engine: EngineTemporal, EngineRef: "run-3"}) + conflictCurrent(t, err, ErrInvalidTransition) +} + +func TestResumeRejectsAForeignPriorExecution(t *testing.T) { + s := newStoreForTest(t) + ctx := context.Background() + a, b := open(t, s, CreateInput{}), open(t, s, CreateInput{}) + if _, _, err := s.AttachExecution(ctx, ExecutionInput{Mutation: as("service:mctl-agent"), WorkItemID: a.ID, Engine: EngineArgo, EngineRef: "own", Phase: PhaseFailed}); err != nil { + t.Fatal(err) + } + foreign, _, err := s.AttachExecution(ctx, ExecutionInput{Mutation: as("service:mctl-agent"), WorkItemID: b.ID, Engine: EngineArgo, EngineRef: "wf", Phase: PhaseSucceeded}) + if err != nil { + t.Fatal(err) + } + _, _, _, err = s.Resume(ctx, ResumeInput{Mutation: as("github:alice"), WorkItemID: a.ID, ExpectedStateVersion: 1, + ResumedFromExecutionID: foreign.ID, Engine: EngineArgo, EngineRef: "wf-2"}) + if !errors.Is(err, ErrInvalid) { + t.Fatalf("err = %v, want ErrInvalid", err) + } +} + +func TestIntentsAreBoundedAndNeverCarrySecrets(t *testing.T) { + s := newStoreForTest(t) + ctx := context.Background() + w := open(t, s, CreateInput{}) + if _, _, err := s.AppendIntent(ctx, IntentInput{Mutation: as("github:alice"), WorkItemID: w.ID, Text: "trailing newline is fine\n"}); err != nil { + t.Fatalf("intent with a trailing newline: %v", err) + } + in := IntentInput{Mutation: as("github:alice"), WorkItemID: w.ID, Text: "please add a retry", Params: []byte(`{"service":"api"}`)} + intent, created, err := s.AppendIntent(ctx, in) + if err != nil || !created || intent.Text != in.Text || intent.ActorPrincipal != "github:alice" { + t.Fatalf("intent = %+v %v %v", intent, created, err) + } + + in.Text = strings.Repeat("x", MaxIntentBytes+1) + if _, _, err := s.AppendIntent(ctx, in); !errors.Is(err, ErrInvalid) { + t.Fatalf("oversized intent: err = %v", err) + } + in.Text = "my token is ghp_" + strings.Repeat("a", 36) + if _, _, err := s.AppendIntent(ctx, in); !errors.Is(err, ErrSecretInText) { + t.Fatalf("secret intent: err = %v", err) + } + in.Text, in.Params = "ok", []byte(`["not","an","object"]`) + if _, _, err := s.AppendIntent(ctx, in); !errors.Is(err, ErrInvalid) { + t.Fatalf("non-object params: err = %v", err) + } + + var stored int + if err := s.pool.QueryRow(ctx, `SELECT count(*) FROM work_item_intents WHERE work_item_id=$1`, w.ID).Scan(&stored); err != nil || stored != 2 { + t.Fatalf("stored intents = %d, %v; rejected ones must not persist", stored, err) + } + events, _ := s.Events(ctx, w.ID) + for _, e := range events { + if strings.Contains(string(e.Detail), "retry") { + t.Fatalf("event %d copies intent text: %s", e.Seq, e.Detail) + } + } +} + +func TestSurfaceRefsCorrelateWithoutCopyingIDsIntoHistory(t *testing.T) { + s := newStoreForTest(t) + ctx := context.Background() + w := open(t, s, CreateInput{}) + m := as("service:mctl-telegram") + m.Surface = "telegram" + in := SurfaceRefInput{Mutation: m, WorkItemID: w.ID, ExternalID: "chat-42", ActorExternalID: "tg-777"} + ref, created, err := s.LinkSurface(ctx, in) + if err != nil || !created || ref.Surface != "telegram" || ref.ActorExternalID != "tg-777" { + t.Fatalf("link = %+v %v %v", ref, created, err) + } + again, created, err := s.LinkSurface(ctx, in) + if err != nil || created || !again.FirstSeenAt.Equal(ref.FirstSeenAt) || again.LastSeenAt.Before(ref.LastSeenAt) { + t.Fatalf("relink = %+v %v %v", again, created, err) + } + events, _ := s.Events(ctx, w.ID) + if len(events) != 2 || events[1].Kind != EventSurfaceLinked { + t.Fatalf("events = %+v", events) + } + if strings.Contains(string(events[1].Detail), "chat-42") || strings.Contains(string(events[1].Detail), "tg-777") { + t.Fatalf("surface_linked copies external ids: %s", events[1].Detail) + } +} + +func TestListDefaultsToOpenAndHidesOthersPrivateItems(t *testing.T) { + s := newStoreForTest(t) + ctx := context.Background() + shared := open(t, s, CreateInput{}) + private := open(t, s, CreateInput{Mutation: as("github:bob"), Visibility: VisibilityPrivate}) + done := open(t, s, CreateInput{}) + open(t, s, CreateInput{Tenant: "other"}) + if _, err := s.Transition(ctx, TransitionInput{Mutation: as("github:alice"), WorkItemID: done.ID, Action: ActionArchive, ExpectedStateVersion: 1}); err != nil { + t.Fatal(err) + } + ids := func(items []WorkItem) map[string]bool { + out := map[string]bool{} + for _, w := range items { + out[w.ID] = true + } + return out + } + + got, err := s.List(ctx, ListFilter{Tenants: []string{"acme"}, Viewer: "github:alice"}) + if err != nil { + t.Fatal(err) + } + if want := map[string]bool{shared.ID: true}; len(got) != 1 || !ids(got)[shared.ID] { + t.Fatalf("alice sees %v, want %v", ids(got), want) + } + got, _ = s.List(ctx, ListFilter{Tenants: []string{"acme"}, Viewer: "github:bob"}) + if len(got) != 2 || !ids(got)[private.ID] { + t.Fatalf("bob sees %v", ids(got)) + } + got, _ = s.List(ctx, ListFilter{Tenants: []string{"acme"}, State: StateArchived}) + if len(got) != 1 || !ids(got)[done.ID] { + t.Fatalf("archived = %v", ids(got)) + } + got, _ = s.List(ctx, ListFilter{AllTenants: true}) + if len(got) != 3 { + t.Fatalf("every tenant, open = %d items, want 3", len(got)) + } + // Tenancy fails closed: a filter naming no tenant matches nothing. + if got, err := s.List(ctx, ListFilter{}); err != nil || len(got) != 0 { + t.Fatalf("zero filter = %d items, %v; want none", len(got), err) + } + // AllTenants lifts the requirement, it does not drop a named tenant. + if got, _ := s.List(ctx, ListFilter{AllTenants: true, Tenants: []string{"other"}}); len(got) != 1 || got[0].Tenant != "other" { + t.Fatalf("admin narrowed to other = %v", got) + } + got, _ = s.List(ctx, ListFilter{Tenants: []string{"acme"}, Owner: "github:bob"}) + if len(got) != 1 || !ids(got)[private.ID] { + t.Fatalf("owner bob = %v", ids(got)) + } + if _, err := s.List(ctx, ListFilter{State: "resumed"}); !errors.Is(err, ErrInvalid) { + t.Fatalf("resumed filter: err = %v", err) + } +} + +func TestCreateRecordsItsKeyEvenWhenExternalKeyDedupes(t *testing.T) { + s := newStoreForTest(t) + ctx := context.Background() + existing := open(t, s, CreateInput{ExternalKey: "https://github.com/mctlhq/mctl-api/issues/1"}) + in := CreateInput{Mutation: keyed("github:bob", "K", "H1"), Tenant: "acme", Visibility: VisibilityTenant, + OriginSurface: "telegram", Title: "same issue", ExternalKey: existing.ExternalKey} + if got, created, err := s.Create(ctx, in); err != nil || created || got.ID != existing.ID { + t.Fatalf("dedupe = %v %v %v", got, created, err) + } + other := in + other.RequestHash = "H2" + if _, _, err := s.Create(ctx, other); !errors.Is(err, ErrIdempotencyKeyReuse) { + t.Fatalf("same key, different request: err = %v", err) + } + if _, err := s.Transition(ctx, TransitionInput{Mutation: as("github:alice"), WorkItemID: existing.ID, Action: ActionComplete, ExpectedStateVersion: 1}); err != nil { + t.Fatal(err) + } + // Retried after the item went terminal, the key still names that item. + if got, created, err := s.Create(ctx, in); err != nil || created || got.ID != existing.ID { + t.Fatalf("retry after terminal = %v %v %v; want %s", got, created, err, existing.ID) + } +} + +func TestTerminalItemsRefuseIntentsAndNewExecutions(t *testing.T) { + s := newStoreForTest(t) + ctx := context.Background() + w := open(t, s, CreateInput{}) + if _, err := s.Transition(ctx, TransitionInput{Mutation: as("github:alice"), WorkItemID: w.ID, Action: ActionArchive, ExpectedStateVersion: 1}); err != nil { + t.Fatal(err) + } + _, _, err := s.AppendIntent(ctx, IntentInput{Mutation: as("github:alice"), WorkItemID: w.ID, Text: "one more thing"}) + conflictCurrent(t, err, ErrInvalidTransition) + _, _, err = s.AttachExecution(ctx, ExecutionInput{Mutation: as("service:mctl-agent"), WorkItemID: w.ID, Engine: EngineArgo, EngineRef: "late", Phase: PhaseRunning}) + conflictCurrent(t, err, ErrInvalidTransition) +} + +func TestAKeyIsBoundToOneOperation(t *testing.T) { + s := newStoreForTest(t) + ctx := context.Background() + w := open(t, s, CreateInput{}) + m := keyed("github:alice", "shared", "same-hash") + if _, err := s.Transition(ctx, TransitionInput{Mutation: m, WorkItemID: w.ID, Action: ActionWait, WaitingReason: WaitingInput, ExpectedStateVersion: 1}); err != nil { + t.Fatal(err) + } + if _, _, err := s.AppendIntent(ctx, IntentInput{Mutation: m, WorkItemID: w.ID, Text: "x"}); !errors.Is(err, ErrIdempotencyKeyReuse) { + t.Fatalf("key reused for another operation: err = %v", err) + } +} + +func TestResumeRefusesAnEngineRefAlreadyRecorded(t *testing.T) { + s := newStoreForTest(t) + ctx := context.Background() + w := open(t, s, CreateInput{}) + if _, _, err := s.AttachExecution(ctx, ExecutionInput{Mutation: as("service:mctl-agent"), WorkItemID: w.ID, Engine: EngineTemporal, EngineRef: "run-1", Phase: PhaseFailed}); err != nil { + t.Fatal(err) + } + _, _, _, err := s.Resume(ctx, ResumeInput{Mutation: as("github:alice"), WorkItemID: w.ID, ExpectedStateVersion: 1, Engine: EngineTemporal, EngineRef: "run-1"}) + if !errors.Is(err, ErrInvalid) { + t.Fatalf("resume onto a recorded engine_ref: err = %v, want ErrInvalid", err) + } +} + +func TestResumeOfAnActiveItemKeepsItActive(t *testing.T) { + s := newStoreForTest(t) + ctx := context.Background() + w := open(t, s, CreateInput{}) + if _, _, err := s.AttachExecution(ctx, ExecutionInput{Mutation: as("service:mctl-agent"), WorkItemID: w.ID, Engine: EngineArgo, EngineRef: "wf-1", Phase: PhaseError}); err != nil { + t.Fatal(err) + } + item, exec, created, err := s.Resume(ctx, ResumeInput{Mutation: as("github:alice"), WorkItemID: w.ID, ExpectedStateVersion: 1, Engine: EngineArgo, EngineRef: "wf-2"}) + if err != nil || !created || item.State != StateActive || item.StateVersion != 2 || exec.Attempt != 2 { + t.Fatalf("resume = %+v %+v %v %v", item, exec, created, err) + } + events, _ := s.Events(ctx, w.ID) + last := events[len(events)-1] + if last.Kind != EventResumed || last.FromState != StateActive || last.ToState != StateActive { + t.Fatalf("last event = %+v", last) + } +} + +func TestSecretsAreRejectedInEveryFreeTextField(t *testing.T) { + s := newStoreForTest(t) + ctx := context.Background() + token := "ghp_" + strings.Repeat("a", 36) + base := CreateInput{Mutation: as("github:alice"), Tenant: "acme", Visibility: VisibilityTenant, OriginSurface: "cli", Title: "t"} + title, key := base, base + title.Title = "use " + token + key.ExternalKey = "https://x:" + token + "@example.com/issue/1" + for name, in := range map[string]CreateInput{"title": title, "external_key": key} { + if _, _, err := s.Create(ctx, in); !errors.Is(err, ErrSecretInText) { + t.Errorf("%s: err = %v, want ErrSecretInText", name, err) + } + } + w := open(t, s, CreateInput{}) + if _, _, err := s.AppendIntent(ctx, IntentInput{Mutation: as("github:alice"), WorkItemID: w.ID, Text: "ok", + Params: []byte(`{"token":"` + token + `"}`)}); !errors.Is(err, ErrSecretInText) { + t.Errorf("params: err = %v, want ErrSecretInText", err) + } + m := as("github:alice") + m.Surface = "telegram" + if _, _, err := s.LinkSurface(ctx, SurfaceRefInput{Mutation: m, WorkItemID: w.ID, ExternalID: token}); !errors.Is(err, ErrSecretInText) { + t.Errorf("external_id: err = %v, want ErrSecretInText", err) + } + var n int + if err := s.pool.QueryRow(ctx, `SELECT count(*) FROM work_items`).Scan(&n); err != nil || n != 1 { + t.Fatalf("work items = %d, %v; rejected creates must not persist", n, err) + } +} + +func TestExternalKeyNeverDedupesOntoWorkTheCallerCouldNotOpen(t *testing.T) { + s := newStoreForTest(t) + ctx := context.Background() + key := "https://github.com/mctlhq/mctl-api/issues/7" + private := open(t, s, CreateInput{Visibility: VisibilityPrivate, ExternalKey: key}) + bob := CreateInput{Mutation: as("github:bob"), Tenant: "acme", Visibility: VisibilityPrivate, OriginSurface: "cli", Title: "x", ExternalKey: key} + if got, _, err := s.Create(ctx, bob); !errors.Is(err, ErrExternalKeyInUse) || got != nil { + t.Fatalf("another member's private item: %v %v", got, err) + } + bob.Visibility = VisibilityTenant + if got, _, err := s.Create(ctx, bob); !errors.Is(err, ErrExternalKeyInUse) || got != nil { + t.Fatalf("tenant request onto a private item: %v %v", got, err) + } + // The owner asking again with the same visibility gets it back. + mine := CreateInput{Mutation: as("github:alice"), Tenant: "acme", Visibility: VisibilityPrivate, OriginSurface: "cli", Title: "x", ExternalKey: key} + if got, created, err := s.Create(ctx, mine); err != nil || created || got.ID != private.ID { + t.Fatalf("owner dedupe = %v %v %v", got, created, err) + } + // A private request never silently lands on a tenant-visible item. + shared := open(t, s, CreateInput{ExternalKey: key + "-shared"}) + mine.ExternalKey = shared.ExternalKey + if _, _, err := s.Create(ctx, mine); !errors.Is(err, ErrExternalKeyInUse) { + t.Fatalf("private request onto a tenant item: err = %v", err) + } +} + +func TestAWaitingItemWithNoExecutionCanStillResume(t *testing.T) { + s := newStoreForTest(t) + ctx := context.Background() + w := open(t, s, CreateInput{}) + waiting, err := s.Transition(ctx, TransitionInput{Mutation: as("github:alice"), WorkItemID: w.ID, Action: ActionWait, WaitingReason: WaitingInput, ExpectedStateVersion: 1}) + if err != nil { + t.Fatal(err) + } + item, exec, _, err := s.Resume(ctx, ResumeInput{Mutation: as("github:alice"), WorkItemID: w.ID, ExpectedStateVersion: waiting.StateVersion, Engine: EngineArgo, EngineRef: "first"}) + if err != nil || item.State != StateActive || exec.Attempt != 1 || exec.ResumedFromExecutionID != "" { + t.Fatalf("resume without a prior execution = %+v %+v %v", item, exec, err) + } + events, _ := s.Events(ctx, w.ID) + if last := events[len(events)-1]; last.Kind != EventResumed || last.FromState != StateWaiting || last.ToState != StateActive { + t.Fatalf("last event = %+v", last) + } + // An active item with nothing before it starts with AttachExecution. + fresh := open(t, s, CreateInput{}) + if _, _, _, err := s.Resume(ctx, ResumeInput{Mutation: as("github:alice"), WorkItemID: fresh.ID, ExpectedStateVersion: 1, Engine: EngineArgo, EngineRef: "x"}); !errors.Is(err, ErrInvalid) { + t.Fatalf("resume of an active item with no execution: err = %v", err) + } +} + +func TestSchemaUpgradesAnEarlierRevision(t *testing.T) { + s := newStoreForTest(t) + ctx := context.Background() + if _, err := s.pool.Exec(ctx, `ALTER TABLE work_item_requests DROP COLUMN actor`); err != nil { + t.Fatal(err) + } + if _, err := s.pool.Exec(ctx, `ALTER TABLE work_item_create_requests DROP COLUMN actor`); err != nil { + t.Fatal(err) + } + again, err := NewStore(ctx, os.Getenv("TEST_DATABASE_URL")) + if err != nil { + t.Fatal(err) + } + defer again.Close() + if _, _, err := again.Create(ctx, CreateInput{Mutation: keyed("github:alice", "k", "h"), Tenant: "acme", Visibility: VisibilityTenant, OriginSurface: "cli", Title: "x"}); err != nil { + t.Fatalf("keyed create after upgrading an old schema: %v", err) + } +} + +func TestAKeyBelongsToOnePrincipal(t *testing.T) { + s := newStoreForTest(t) + ctx := context.Background() + in := CreateInput{Mutation: keyed("github:alice", "shared-key", "same-body"), Tenant: "acme", Visibility: VisibilityTenant, OriginSurface: "cli", Title: "x"} + w, _, err := s.Create(ctx, in) + if err != nil { + t.Fatal(err) + } + in.Mutation = keyed("github:bob", "shared-key", "same-body") + if _, _, err := s.Create(ctx, in); !errors.Is(err, ErrIdempotencyKeyReuse) { + t.Fatalf("bob replaying alice's create key: err = %v", err) + } + tr := TransitionInput{Mutation: keyed("github:alice", "t", "h"), WorkItemID: w.ID, Action: ActionArchive, ExpectedStateVersion: 1} + if _, err := s.Transition(ctx, tr); err != nil { + t.Fatal(err) + } + tr.Mutation = keyed("github:bob", "t", "h") + if _, err := s.Transition(ctx, tr); !errors.Is(err, ErrIdempotencyKeyReuse) { + t.Fatalf("bob replaying alice's transition key: err = %v", err) + } +} + +func TestPaddedIdentitiesAreRefused(t *testing.T) { + s := newStoreForTest(t) + for name, in := range map[string]CreateInput{ + "actor": {Mutation: as(" github:alice "), Tenant: "acme", Visibility: VisibilityTenant, OriginSurface: "cli", Title: "x"}, + "tenant": {Mutation: as("github:alice"), Tenant: "acme ", Visibility: VisibilityTenant, OriginSurface: "cli", Title: "x"}, + } { + if _, _, err := s.Create(context.Background(), in); !errors.Is(err, ErrInvalid) { + t.Errorf("%s: err = %v, want ErrInvalid", name, err) + } + } +} diff --git a/internal/workitems/types.go b/internal/workitems/types.go new file mode 100644 index 00000000..effe8dd3 --- /dev/null +++ b/internal/workitems/types.go @@ -0,0 +1,243 @@ +// Package workitems is the durable owner of "the piece of work a human asked +// for" (docs/work-context-contract.md, workitem/v1). Surfaces are adapters: +// they correlate to a WorkItem, they never hold its lifecycle. +// +// The package owns the schema, the transition table and the store. Who may +// call it, and as whom, is decided by the HTTP layer from authentication; the +// store records the principal it is handed and never derives one from a +// request body. +package workitems + +import ( + "encoding/json" + "errors" + "fmt" + "regexp" + "time" +) + +// SchemaVersion labels every work-item payload. A breaking change ships as a +// new label, never as a silent change to this one. +const SchemaVersion = "workitem/v1" + +// ID prefixes (contract "ID scheme"). +const ( + WorkItemIDPrefix = "wi_" + ExecutionIDPrefix = "we_" +) + +// States. `resumed` is deliberately not one: it is only ever an event kind. +const ( + StateActive = "active" + StateWaiting = "waiting" + StateCompleted = "completed" + StateSuperseded = "superseded" + StateArchived = "archived" +) + +// FilterOpen is the virtual list filter for every non-terminal state, the +// default when a list names no state (mirrors alerts.StatusActive). +const FilterOpen = "open" + +// Waiting reasons. +const ( + WaitingInput = "input" + WaitingApproval = "approval" +) + +// Visibility. +const ( + VisibilityTenant = "tenant" + VisibilityPrivate = "private" +) + +// Lifecycle actions accepted by Transition. +const ( + ActionWait = "wait" + ActionResume = "resume" + ActionComplete = "complete" + ActionSupersede = "supersede" + ActionArchive = "archive" +) + +// Event kinds recorded by this slice. The contract also names +// approval_requested and approval_decided; they arrive with the approval +// projection. +const ( + EventCreated = "created" + EventStateChanged = "state_changed" + EventResumed = "resumed" + EventIntentAppended = "intent_appended" + EventExecutionAttached = "execution_attached" + EventSurfaceLinked = "surface_linked" +) + +// Execution engines and phases. +const ( + EngineTemporal = "temporal" + EngineArgo = "argo" + + PhasePending = "Pending" + PhaseRunning = "Running" + PhaseSucceeded = "Succeeded" + PhaseFailed = "Failed" + PhaseError = "Error" +) + +// Bounds. Intent text is capped by the contract; the rest keep a correlation +// field from turning into a place to park content. +const ( + MaxIntentBytes = 8 << 10 + MaxIntentParamBytes = 8 << 10 + MaxTitleBytes = 256 + MaxKeyBytes = 512 + MaxExternalIDBytes = 256 + MaxEngineRefBytes = 256 +) + +// Transitions is the single enforcement point for lifecycle rules: from +// state, action -> to state. Anything not listed, including every move out of +// a terminal state, is rejected. +var Transitions = map[string]map[string]string{ + StateActive: { + ActionWait: StateWaiting, + ActionComplete: StateCompleted, + ActionSupersede: StateSuperseded, + ActionArchive: StateArchived, + }, + StateWaiting: { + ActionResume: StateActive, + ActionComplete: StateCompleted, + ActionSupersede: StateSuperseded, + ActionArchive: StateArchived, + }, +} + +// Next returns the state action leads to from `from`, or +// ErrInvalidTransition. +func Next(from, action string) (string, error) { + if to, ok := Transitions[from][action]; ok { + return to, nil + } + return "", fmt.Errorf("%w: %s from %s", ErrInvalidTransition, action, from) +} + +// IsTerminal reports whether no transition leaves state. +func IsTerminal(state string) bool { + _, ok := Transitions[state] + return !ok +} + +// IsTerminalPhase reports whether an execution in phase has finished. +func IsTerminalPhase(phase string) bool { + return phase == PhaseSucceeded || phase == PhaseFailed || phase == PhaseError +} + +func validPhase(phase string) bool { + return phase == PhasePending || phase == PhaseRunning || IsTerminalPhase(phase) +} + +func validState(state string) bool { + switch state { + case StateActive, StateWaiting, StateCompleted, StateSuperseded, StateArchived: + return true + } + return false +} + +// surfacePattern bounds a surface label: a short lowercase token such as +// "telegram", "mcp", "cli" or "web". It is a label, never an identity. +var surfacePattern = regexp.MustCompile(`^[a-z][a-z0-9_-]{0,31}$`) + +// Sentinel errors. The HTTP layer maps each to one status and typed reason. +var ( + ErrNotFound = errors.New("work item not found") + ErrInvalid = errors.New("invalid work item request") + ErrInvalidTransition = errors.New("transition not allowed") + ErrVersionConflict = errors.New("state_version does not match") + ErrExecutionActive = errors.New("work item already has a non-terminal execution") + ErrIdempotencyKeyReuse = errors.New("idempotency key was already used for a different request") + ErrSecretInText = errors.New("text matches a secret pattern") + ErrExternalKeyInUse = errors.New("external_key already names open work this request cannot share") +) + +// ConflictError carries the item as it is now, so a 409 can tell the caller +// the current state and version instead of making it re-read. +type ConflictError struct { + Err error + Current *WorkItem +} + +func (e *ConflictError) Error() string { return e.Err.Error() } +func (e *ConflictError) Unwrap() error { return e.Err } + +// WorkItem is one piece of work a human asked for. +type WorkItem struct { + ID string `json:"id"` + Tenant string `json:"tenant"` + OwnerPrincipal string `json:"owner_principal"` + Visibility string `json:"visibility"` + OriginSurface string `json:"origin_surface"` + Title string `json:"title"` + ExternalKey string `json:"external_key,omitempty"` + State string `json:"state"` + WaitingReason string `json:"waiting_reason,omitempty"` + SupersededBy string `json:"superseded_by,omitempty"` + StateVersion int64 `json:"state_version"` + CreatedBy string `json:"created_by"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` + CompletedAt *time.Time `json:"completed_at,omitempty"` + SchemaVersion string `json:"schema_version"` +} + +// Event is one append-only lifecycle record. +type Event struct { + WorkItemID string `json:"work_item_id"` + Seq int64 `json:"seq"` + Kind string `json:"kind"` + FromState string `json:"from_state,omitempty"` + ToState string `json:"to_state,omitempty"` + ActorPrincipal string `json:"actor_principal"` + Surface string `json:"surface,omitempty"` + RequestID string `json:"request_id,omitempty"` + Detail json.RawMessage `json:"detail,omitempty"` + CreatedAt time.Time `json:"created_at"` +} + +// Intent is a bounded statement of what was asked for. Never a transcript. +type Intent struct { + ID int64 `json:"id"` + WorkItemID string `json:"work_item_id"` + ActorPrincipal string `json:"actor_principal"` + Surface string `json:"surface,omitempty"` + Text string `json:"text"` + Params json.RawMessage `json:"params,omitempty"` + CreatedAt time.Time `json:"created_at"` +} + +// Execution correlates the work item to one engine run. The engine keeps the +// execution detail; this is only the reference to it. +type Execution struct { + ID string `json:"id"` + WorkItemID string `json:"work_item_id"` + Engine string `json:"engine"` + EngineRef string `json:"engine_ref"` + Attempt int `json:"attempt"` + ResumedFromExecutionID string `json:"resumed_from_execution_id,omitempty"` + Phase string `json:"phase"` + StartedAt time.Time `json:"started_at"` + EndedAt *time.Time `json:"ended_at,omitempty"` +} + +// SurfaceRef correlates a work item to a surface-native conversation for +// reply routing. ActorExternalID is correlation metadata, never identity: it +// resolves to a principal only through a SurfaceIdentityLink (mctl-api#350). +type SurfaceRef struct { + WorkItemID string `json:"work_item_id"` + Surface string `json:"surface"` + ExternalID string `json:"external_id"` + ActorExternalID string `json:"actor_external_id,omitempty"` + FirstSeenAt time.Time `json:"first_seen_at"` + LastSeenAt time.Time `json:"last_seen_at"` +}