tasks/pkg/tasks/engine.go

// Copyright © 2026 Hanzo AI. MIT License.

package tasks

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"log/slog"
	"strings"
	"sync"
	"time"

	"github.com/hanzoai/tasks/pkg/sdk/temporal"
	storepkg "github.com/hanzoai/tasks/pkg/tasks/store"
	"github.com/robfig/cron/v3"
)

// cancelTimeout bounds how long a workflow may sit in CANCELING before
// the sweeper auto-acks it to CANCELED. Mutable for tests.
var cancelTimeout = 60 * time.Second

// cancelTracker records when each (ns, wfID, runID) entered CANCELING.
type cancelTracker struct {
	mu sync.Mutex
	at map[string]time.Time
}

func newCancelTracker() *cancelTracker { return &cancelTracker{at: map[string]time.Time{}} }

func (c *cancelTracker) mark(ns, wfID, runID string) {
	c.mu.Lock()
	c.at[wfKey(ns, wfID, runID)] = time.Now()
	c.mu.Unlock()
}

func (c *cancelTracker) clear(ns, wfID, runID string) {
	c.mu.Lock()
	delete(c.at, wfKey(ns, wfID, runID))
	c.mu.Unlock()
}

// expired returns the keys older than cancelTimeout. Caller must hold no lock.
func (c *cancelTracker) expired() []string {
	c.mu.Lock()
	defer c.mu.Unlock()
	cutoff := time.Now().Add(-cancelTimeout)
	out := []string{}
	for k, t := range c.at {
		if t.Before(cutoff) {
			out = append(out, k)
		}
	}
	return out
}

func jsonUnmarshal(b []byte, v any) error { return json.Unmarshal(b, v) }

// engine wraps the store with domain operations. ZAP handlers and the
// HTTP shim both call into this layer — there is exactly one
// implementation per opcode.
//
// Workflow execution semantics (deterministic replay of user code,
// activity heartbeating, child-workflow lifecycle) are NOT implemented
// here yet. StartWorkflow records the execution as RUNNING and emits
// no events; SignalWorkflow / CancelWorkflow / TerminateWorkflow apply
// state transitions on the record only. This is enough for the UI to
// be inspectable end-to-end and for higher-level callers to depend on
// the API while the worker SDK runtime lands.
type engine struct {
	store      *store
	broker     *broker
	disp       *dispatcher
	cancelling *cancelTracker
	workers    *workerRegistry
	runMu      *keyedMutex // serializes advancing a single (org, ns, wf, run)
	principal  Principal   // zero = root tenant (embedded/dev). Its org is stamped on every emitted Event.
	// log reports what the background sweepers cannot return to a caller.
	// Always non-nil (newEngine defaults it, As carries it, Embed overrides
	// it with EmbedConfig.Logger), so no call site needs a guard.
	log *slog.Logger
	// fails throttles failure reporting — cron schedules here, activities
	// and workflows in failure.go, one tracker because there is one throttle
	// rule. Shared across tenant views like broker/disp/workers, because the
	// root sweeper fires every tenant's schedules through a throwaway view
	// and the count must follow the work, not the view; keys are
	// principal-qualified for the same reason.
	fails *failTracker
}

func newEngine(s *store) *engine {
	return &engine{store: s, broker: newBroker(), disp: newDispatcher(), cancelling: newCancelTracker(), workers: newWorkerRegistry(), runMu: newKeyedMutex(), log: slog.Default(), fails: newFailTracker(failReportThrottle)}
}

// As returns an engine view scoped to the tenant p. Reads and writes route
// to that tenant's own per-namespace shard directory (store.as); the
// dispatcher/broker/worker registry stay SHARED, so a tenant's workflows
// dispatch to whatever worker is subscribed on (ns, queue) regardless of
// which view enqueued them. The zero principal returns e unchanged (the
// root / embedded view).
func (e *engine) As(p Principal) *engine {
	if p == e.principal {
		return e
	}
	return &engine{store: e.store.as(p), broker: e.broker, disp: e.disp, cancelling: e.cancelling, workers: e.workers, runMu: e.runMu, principal: p, log: e.log, fails: e.fails}
}

// emit publishes an event tagged with the engine's org so per-org SSE
// filtering can trust OrgID. Use everywhere instead of e.broker.publish.
func (e *engine) emit(ev Event) {
	ev.OrgID = e.principal.Org
	e.broker.publish(ev)
}

// ── ids ─────────────────────────────────────────────────────────────

func newRunId() string {
	var b [16]byte
	_, _ = rand.Read(b[:])
	return hex.EncodeToString(b[:])
}

// ── namespaces ──────────────────────────────────────────────────────

func (e *engine) RegisterNamespace(ns Namespace) error {
	if ns.NamespaceInfo.Name == "" {
		return fmt.Errorf("namespace name required")
	}
	if ns.NamespaceInfo.State == "" {
		ns.NamespaceInfo.State = "NAMESPACE_STATE_REGISTERED"
	}
	if ns.NamespaceInfo.CreateTime == "" {
		ns.NamespaceInfo.CreateTime = nowRFC3339()
	}
	if ns.Config.WorkflowExecutionRetentionTtl == "" {
		ns.Config.WorkflowExecutionRetentionTtl = "720h"
	}
	if ns.Config.APSLimit == 0 {
		ns.Config.APSLimit = 400
	}
	if ns.NamespaceInfo.Region == "" {
		ns.NamespaceInfo.Region = "embedded"
	}
	ns.IsActive = ns.NamespaceInfo.State == "NAMESPACE_STATE_REGISTERED"
	if err := e.store.put("ns/"+ns.NamespaceInfo.Name, ns); err != nil {
		return err
	}
	e.emit(Event{Kind: "namespace.registered", Namespace: ns.NamespaceInfo.Name, Data: ns})
	return nil
}

// namespace returns ns's registry row, registering it on first use. A
// namespace is an addressing fact, not a resource to provision: the shard
// it names is created on demand (store.Manager.Get), so its registry row
// is too. This is the ONE place a namespace comes into existence, and
// every operation that needs one calls it instead of asking whether it
// already exists.
func (e *engine) namespace(name string) (*Namespace, error) {
	if err := storepkg.ValidName(name); err != nil {
		return nil, err
	}
	if n, ok, err := e.DescribeNamespace(name); err != nil {
		return nil, err
	} else if ok {
		return n, nil
	}
	if err := e.RegisterNamespace(Namespace{NamespaceInfo: NamespaceInfo{Name: name}}); err != nil {
		return nil, err
	}
	n, ok, err := e.DescribeNamespace(name)
	if err != nil {
		return nil, err
	}
	if !ok {
		return nil, fmt.Errorf("namespace %q vanished after registration", name)
	}
	return n, nil
}

func (e *engine) DescribeNamespace(name string) (*Namespace, bool, error) {
	var n Namespace
	ok, err := e.store.get("ns/"+name, &n)
	if !ok {
		return nil, false, err
	}
	return &n, true, err
}

func (e *engine) ListNamespaces() ([]Namespace, error) {
	return listInto[Namespace](e.store, "ns/")
}

// ── workflows ───────────────────────────────────────────────────────

func (e *engine) StartWorkflow(ns, workflowId, runId string, typ TypeRef, taskQueue string, input any) (*WorkflowExecution, error) {
	return e.startWorkflowFull(ns, workflowId, runId, typ, taskQueue, input, "", nil, nil, "")
}

// StartWorkflowWithRequestID is the idempotent form. When requestID is
// non-empty and a prior Start under the same (workflowID, requestID)
// landed, the existing execution is returned without a new run.
func (e *engine) StartWorkflowWithRequestID(ns, workflowId, runId string, typ TypeRef, taskQueue string, input any, requestID string) (*WorkflowExecution, error) {
	return e.startWorkflowFull(ns, workflowId, runId, typ, taskQueue, input, requestID, nil, nil, "")
}

func (e *engine) startWorkflowWithRequestID(ns, workflowId, runId string, typ TypeRef, taskQueue string, input any, requestID string) (*WorkflowExecution, error) {
	return e.startWorkflowFull(ns, workflowId, runId, typ, taskQueue, input, requestID, nil, nil, "")
}

// startWorkflowFull is the single start implementation. searchAttrs/memo
// are stored on the execution (searchAttrs are visibility-queryable — see
// fieldValue). conflictPolicy governs a live execution already existing
// under workflowId (see resolveWorkflowIdConflict).
func (e *engine) startWorkflowFull(ns, workflowId, runId string, typ TypeRef, taskQueue string, input any, requestID string, searchAttrs map[string]any, memo any, conflictPolicy string) (*WorkflowExecution, error) {
	if _, err := e.namespace(ns); err != nil {
		return nil, err
	}
	if typ.Name == "" {
		return nil, fmt.Errorf("workflow type required")
	}
	if workflowId == "" {
		workflowId = strings.ToLower(typ.Name) + "-" + newRunId()[:8]
	}
	if requestID != "" {
		if existingRun, ok, _ := e.lookupIdempotency(ns, workflowId, requestID); ok {
			if wf, ok2, _ := e.DescribeWorkflow(ns, workflowId, existingRun); ok2 {
				return wf, nil
			}
		}
	}
	// Workflow-id reuse policy: an existing NON-terminal run under
	// workflowId is either reused, terminated, or rejected before a fresh
	// run is created. Default (empty) preserves the historical behaviour
	// (a new run is created under the same workflowId).
	if existing, handled, err := e.resolveWorkflowIdConflict(ns, workflowId, conflictPolicy); err != nil {
		return nil, err
	} else if handled {
		return existing, nil
	}
	if runId == "" {
		runId = newRunId()
	}
	if taskQueue == "" {
		taskQueue = "default"
	}
	startTime := nowRFC3339()
	wf := WorkflowExecution{
		Execution:   ExecutionRef{WorkflowId: workflowId, RunId: runId},
		Type:        typ,
		StartTime:   startTime,
		Status:      "WORKFLOW_EXECUTION_STATUS_RUNNING",
		TaskQueue:   taskQueue,
		Input:       input,
		SearchAttrs: searchAttrs,
		Memo:        memo,
	}
	key := fmt.Sprintf("wf/%s/%s/%s", ns, workflowId, runId)
	if err := e.store.put(key, wf); err != nil {
		return nil, err
	}
	if requestID != "" {
		_ = e.store.put(fmt.Sprintf("idem/%s/%s/%s", ns, workflowId, requestID), runId)
	}
	if _, err := e.appendHistory(ns, workflowId, runId, "WORKFLOW_EXECUTION_STARTED", map[string]any{
		"workflowType": typ.Name,
		"taskQueue":    taskQueue,
		"input":        input,
	}); err != nil {
		return nil, err
	}
	// Reload to surface bumped HistoryLen.
	if reloaded, ok, _ := e.DescribeWorkflow(ns, workflowId, runId); ok {
		wf = *reloaded
	}
	// Deliver the first workflow task carrying the run's full history
	// (WORKFLOW_EXECUTION_STARTED); the worker replays it from event 0.
	if err := e.scheduleWorkflowTask(ns, workflowId, runId); err != nil {
		return nil, err
	}
	e.emit(Event{
		Kind:       "workflow.started",
		Namespace:  ns,
		WorkflowID: workflowId,
		RunID:      runId,
		Data:       wf,
	})
	return &wf, nil
}

// resolveWorkflowIdConflict applies the workflow-id reuse policy against a
// live (non-terminal) run already under workflowId. Returns (existing,
// true) when the caller should short-circuit and reuse that run,
// (nil,false) to proceed with a fresh run, or an error for FAIL. An empty
// policy means "no policy" — proceed (historical behaviour). Recognised
// policies (long WORKFLOW_ID_CONFLICT_POLICY_* or the short suffix):
//   - USE_EXISTING       → reuse the live run
//   - TERMINATE_EXISTING → terminate the live run, then start fresh
//   - FAIL               → error if a live run exists
func (e *engine) resolveWorkflowIdConflict(ns, workflowId, policy string) (*WorkflowExecution, bool, error) {
	if policy == "" {
		return nil, false, nil
	}
	existing, ok, _ := e.DescribeWorkflow(ns, workflowId, "")
	if !ok || isTerminal(existing.Status) {
		return nil, false, nil
	}
	switch normalizeConflictPolicy(policy) {
	case "USE_EXISTING":
		return existing, true, nil
	case "TERMINATE_EXISTING":
		if _, err := e.TerminateWorkflowWithReason(ns, existing.Execution.WorkflowId, existing.Execution.RunId, "superseded by workflow-id conflict policy", "engine"); err != nil {
			return nil, false, err
		}
		return nil, false, nil
	case "FAIL":
		return nil, false, fmt.Errorf("workflow %q already running (conflict policy FAIL)", workflowId)
	}
	return nil, false, nil
}

func normalizeConflictPolicy(p string) string {
	p = strings.ToUpper(p)
	p = strings.TrimPrefix(p, "WORKFLOW_ID_CONFLICT_POLICY_")
	return p
}

// lookupIdempotency returns the runID previously associated with
// (workflowID, requestID), if any.
func (e *engine) lookupIdempotency(ns, workflowID, requestID string) (string, bool, error) {
	var runID string
	ok, err := e.store.get(fmt.Sprintf("idem/%s/%s/%s", ns, workflowID, requestID), &runID)
	return runID, ok, err
}

func (e *engine) DescribeWorkflow(ns, workflowId, runId string) (*WorkflowExecution, bool, error) {
	if runId != "" {
		var wf WorkflowExecution
		ok, err := e.store.get(fmt.Sprintf("wf/%s/%s/%s", ns, workflowId, runId), &wf)
		if !ok {
			return nil, false, err
		}
		return &wf, true, err
	}
	// runId not supplied — return the CURRENT execution under workflowId: the
	// open (non-terminal) run if one exists, else the most recent. This makes
	// the successor of a continueAsNew (a fresh RUNNING run that may share the
	// closed run's second-granularity StartTime) the one resolved, so
	// signal / describe target the live run.
	rows, err := listInto[WorkflowExecution](e.store, fmt.Sprintf("wf/%s/%s/", ns, workflowId))
	if err != nil || len(rows) == 0 {
		return nil, false, err
	}
	best := 0
	for i := 1; i < len(rows); i++ {
		if moreCurrentRun(rows[i], rows[best]) {
			best = i
		}
	}
	return &rows[best], true, nil
}

// moreCurrentRun reports whether run a is a "more current" execution than b
// for the same workflowId: an open run outranks a terminal one; a
// CONTINUED_AS_NEW run (which always has a successor) is the least current, so
// the completed tail of a continue-as-new chain resolves over its closed
// predecessor even when they share a second-granularity StartTime; otherwise
// the later StartTime wins, with the runId as a deterministic tiebreak.
func moreCurrentRun(a, b WorkflowExecution) bool {
	aTerm, bTerm := isTerminal(a.Status), isTerminal(b.Status)
	if aTerm != bTerm {
		return !aTerm
	}
	const can = "WORKFLOW_EXECUTION_STATUS_CONTINUED_AS_NEW"
	if aCAN, bCAN := a.Status == can, b.Status == can; aCAN != bCAN {
		return !aCAN
	}
	if a.StartTime != b.StartTime {
		return a.StartTime > b.StartTime
	}
	return a.Execution.RunId > b.Execution.RunId
}

func (e *engine) ListWorkflows(ns string) ([]WorkflowExecution, error) {
	return listInto[WorkflowExecution](e.store, fmt.Sprintf("wf/%s/", ns))
}

func (e *engine) terminalTransition(ns, workflowId, runId, status, evKind, eventType string, attrs map[string]any) (*WorkflowExecution, error) {
	wf, ok, err := e.DescribeWorkflow(ns, workflowId, runId)
	if !ok {
		return nil, fmt.Errorf("workflow %s/%s/%s not found", ns, workflowId, runId)
	}
	if err != nil {
		return nil, err
	}
	if isTerminal(wf.Status) {
		return wf, nil
	}
	wf.Status = status
	wf.CloseTime = nowRFC3339()
	if err := e.store.put(fmt.Sprintf("wf/%s/%s/%s", ns, wf.Execution.WorkflowId, wf.Execution.RunId), wf); err != nil {
		return nil, err
	}
	if _, err := e.appendHistory(ns, wf.Execution.WorkflowId, wf.Execution.RunId, eventType, attrs); err != nil {
		return nil, err
	}
	// Reload to capture the bumped HistoryLen.
	wf, _, _ = e.DescribeWorkflow(ns, wf.Execution.WorkflowId, wf.Execution.RunId)
	e.emit(Event{
		Kind:       evKind,
		Namespace:  ns,
		WorkflowID: wf.Execution.WorkflowId,
		RunID:      wf.Execution.RunId,
		Data:       wf,
	})
	// The other half of the silent path. An activity-level report cannot see
	// a decider that faults before it schedules anything (bad input, a panic
	// on episode 0): that workflow fails with no activity record to speak for
	// it, which is the same class of bug as the 11-day incident with an even
	// thinner trail. CANCELED/TERMINATED are operator actions, not faults, so
	// they neither record nor clear.
	switch status {
	case "WORKFLOW_EXECUTION_STATUS_FAILED":
		e.recordFailure(workflowFailureIdentity(ns, wf), false, workflowFailureText(attrs))
	case "WORKFLOW_EXECUTION_STATUS_COMPLETED":
		e.clearFailure(workflowFailureIdentity(ns, wf))
	}
	return wf, nil
}

// workflowFailureText renders the failure a terminal transition recorded into
// history. attrs["failure"] carries the encoded envelope the decider sent;
// decoding it yields the message a human reads, and temporal.Decode is
// fail-secure on hostile or absent bytes.
func workflowFailureText(attrs map[string]any) string {
	s, _ := attrs["failure"].(string)
	if s == "" {
		return ""
	}
	return temporal.Decode([]byte(s)).Error()
}

func isTerminal(status string) bool {
	switch status {
	case "WORKFLOW_EXECUTION_STATUS_COMPLETED",
		"WORKFLOW_EXECUTION_STATUS_FAILED",
		"WORKFLOW_EXECUTION_STATUS_CANCELED",
		"WORKFLOW_EXECUTION_STATUS_TERMINATED",
		"WORKFLOW_EXECUTION_STATUS_TIMED_OUT",
		"WORKFLOW_EXECUTION_STATUS_CONTINUED_AS_NEW":
		return true
	}
	return false
}

func (e *engine) CancelWorkflow(ns, workflowId, runId string) (*WorkflowExecution, error) {
	return e.CancelWorkflowWithReason(ns, workflowId, runId, "", "")
}

// CancelWorkflowWithReason records WORKFLOW_EXECUTION_CANCEL_REQUESTED
// and transitions the workflow to CANCELING. If a worker is subscribed
// to the workflow's task queue, OpcodeDeliverCancelRequest is pushed
// and the engine waits for the worker to ack via a Canceled command on
// RespondWorkflowTaskCompleted. If no worker is subscribed, the cancel
// is auto-acked synchronously to CANCELED so the API converges in
// the absence of a cooperating worker.
func (e *engine) CancelWorkflowWithReason(ns, workflowId, runId, reason, identity string) (*WorkflowExecution, error) {
	wf, ok, err := e.DescribeWorkflow(ns, workflowId, runId)
	if !ok {
		return nil, fmt.Errorf("workflow %s/%s/%s not found", ns, workflowId, runId)
	}
	if err != nil {
		return nil, err
	}
	if isTerminal(wf.Status) {
		return wf, nil
	}
	if wf.Status == "WORKFLOW_EXECUTION_STATUS_CANCELING" {
		// Re-cancel while CANCELING: push again to subscribers but do
		// not append a second CANCEL_REQUESTED.
		if e.disp != nil {
			e.disp.PushCancelRequest(ns, wf.TaskQueue, wf.Execution.WorkflowId, wf.Execution.RunId, reason, identity)
		}
		return wf, nil
	}
	if _, err := e.appendHistory(ns, workflowId, runId, "WORKFLOW_EXECUTION_CANCEL_REQUESTED", map[string]any{
		"reason":   reason,
		"identity": identity,
	}); err != nil {
		return nil, err
	}
	wf, _, _ = e.DescribeWorkflow(ns, workflowId, runId)
	subscribed := e.disp != nil && e.disp.HasSubscribers(ns, wf.TaskQueue, kindWorkflow)
	if !subscribed {
		// No worker — auto-ack to CANCELED synchronously.
		return e.terminalTransition(ns, workflowId, runId, "WORKFLOW_EXECUTION_STATUS_CANCELED", "workflow.canceled", "WORKFLOW_EXECUTION_CANCELED", map[string]any{
			"reason":   reason,
			"identity": identity,
		})
	}
	wf.Status = "WORKFLOW_EXECUTION_STATUS_CANCELING"
	if err := e.store.put(fmt.Sprintf("wf/%s/%s/%s", ns, wf.Execution.WorkflowId, wf.Execution.RunId), wf); err != nil {
		return nil, err
	}
	if e.cancelling != nil {
		e.cancelling.mark(ns, wf.Execution.WorkflowId, wf.Execution.RunId)
	}
	e.disp.PushCancelRequest(ns, wf.TaskQueue, wf.Execution.WorkflowId, wf.Execution.RunId, reason, identity)
	e.emit(Event{
		Kind:       "workflow.cancel_requested",
		Namespace:  ns,
		WorkflowID: wf.Execution.WorkflowId,
		RunID:      wf.Execution.RunId,
		Data:       wf,
	})
	return wf, nil
}

// AckCanceled finalizes a CANCELING workflow → CANCELED. Idempotent
// against terminal states; called from the respond-workflow-task ack
// path and from the cancel sweeper.
func (e *engine) AckCanceled(ns, workflowID, runID, reason, identity string) (*WorkflowExecution, error) {
	if e.cancelling != nil {
		e.cancelling.clear(ns, workflowID, runID)
	}
	return e.terminalTransition(ns, workflowID, runID, "WORKFLOW_EXECUTION_STATUS_CANCELED", "workflow.canceled", "WORKFLOW_EXECUTION_CANCELED", map[string]any{
		"reason":   reason,
		"identity": identity,
	})
}

// sweepCanceling auto-acks workflows that have been CANCELING longer
// than cancelTimeout. Called periodically by runScheduler.
func (e *engine) sweepCanceling() {
	if e.cancelling == nil {
		return
	}
	for _, k := range e.cancelling.expired() {
		parts := strings.SplitN(k, "|", 3)
		if len(parts) != 3 {
			e.cancelling.mu.Lock()
			delete(e.cancelling.at, k)
			e.cancelling.mu.Unlock()
			continue
		}
		_, _ = e.AckCanceled(parts[0], parts[1], parts[2], "sweep:cancel-timeout", "engine")
	}
}

func (e *engine) TerminateWorkflow(ns, workflowId, runId string) (*WorkflowExecution, error) {
	return e.TerminateWorkflowWithReason(ns, workflowId, runId, "", "")
}

func (e *engine) TerminateWorkflowWithReason(ns, workflowId, runId, reason, identity string) (*WorkflowExecution, error) {
	return e.terminalTransition(ns, workflowId, runId, "WORKFLOW_EXECUTION_STATUS_TERMINATED", "workflow.terminated", "WORKFLOW_EXECUTION_TERMINATED", map[string]any{
		"reason":   reason,
		"identity": identity,
	})
}

func (e *engine) SignalWorkflow(ns, workflowId, runId, name string, payload any) error {
	return e.signalWorkflow(ns, workflowId, runId, name, payload, "")
}

func (e *engine) signalWorkflow(ns, workflowId, runId, name string, payload any, identity string) error {
	wf, ok, err := e.DescribeWorkflow(ns, workflowId, runId)
	if !ok || err != nil {
		return fmt.Errorf("workflow not found")
	}
	if isTerminal(wf.Status) {
		return fmt.Errorf("workflow not running: status=%s", wf.Status)
	}
	// Resolve to the concrete run so the signal event and the re-dispatched
	// workflow task both target the same run (DescribeWorkflow with an empty
	// runId returns the latest execution).
	wfID, run := wf.Execution.WorkflowId, wf.Execution.RunId
	// Serialize with the run's other advancers: append SIGNALED then
	// schedule a workflow task so the stateless worker replays the extended
	// history and delivers the signal to its handler. Without the
	// re-dispatch a signal-driven workflow (condition on a signal-populated
	// queue) would never wake until some other event advanced the run.
	unlock := e.lockRun(ns, wfID, run)
	defer unlock()
	if _, err := e.appendHistory(ns, wfID, run, "WORKFLOW_EXECUTION_SIGNALED", map[string]any{
		"signalName": name,
		"input":      payload,
		"identity":   identity,
	}); err != nil {
		return err
	}
	wf, _, _ = e.DescribeWorkflow(ns, wfID, run)
	e.emit(Event{
		Kind:       "workflow.signaled",
		Namespace:  ns,
		WorkflowID: wfID,
		RunID:      run,
		Data:       map[string]any{"signal": name, "input": payload, "history_length": wf.HistoryLen},
	})
	return e.scheduleWorkflowTask(ns, wfID, run)
}

// SignalWithStartWorkflow signals an existing running workflow under
// workflowId; if none is running, starts a fresh one and signals that.
// The (start, signal) pair is atomic from the caller's view.
func (e *engine) SignalWithStartWorkflow(ns, workflowId, runId string, typ TypeRef, taskQueue string, input any, signalName string, signalPayload any, requestID string) (*WorkflowExecution, error) {
	return e.signalWithStartFull(ns, workflowId, runId, typ, taskQueue, input, signalName, signalPayload, requestID, nil, nil)
}

// signalWithStartFull is the search-attribute-aware form: on the
// fresh-start branch the execution is created with searchAttrs/memo (so a
// signalWithStart-created run is visibility-queryable just like a Start).
// USE_EXISTING semantics: a live run under workflowId is signalled, never
// duplicated.
func (e *engine) signalWithStartFull(ns, workflowId, runId string, typ TypeRef, taskQueue string, input any, signalName string, signalPayload any, requestID string, searchAttrs map[string]any, memo any) (*WorkflowExecution, error) {
	if existing, ok, _ := e.DescribeWorkflow(ns, workflowId, ""); ok && !isTerminal(existing.Status) {
		if err := e.SignalWorkflow(ns, existing.Execution.WorkflowId, existing.Execution.RunId, signalName, signalPayload); err != nil {
			return nil, err
		}
		return existing, nil
	}
	wf, err := e.startWorkflowFull(ns, workflowId, runId, typ, taskQueue, input, requestID, searchAttrs, memo, "")
	if err != nil {
		return nil, err
	}
	if err := e.SignalWorkflow(ns, wf.Execution.WorkflowId, wf.Execution.RunId, signalName, signalPayload); err != nil {
		return nil, err
	}
	wf, _, _ = e.DescribeWorkflow(ns, wf.Execution.WorkflowId, wf.Execution.RunId)
	return wf, nil
}

// ── history ────────────────────────────────────────────────────────

// appendHistory writes a HistoryEvent to durable storage and bumps the
// workflow's HistoryLen. EventId is monotonic per (ns, workflowID,
// runID). Returns the appended event.
func (e *engine) appendHistory(ns, workflowID, runID, eventType string, attrs map[string]any) (*HistoryEvent, error) {
	wf, ok, err := e.DescribeWorkflow(ns, workflowID, runID)
	if !ok {
		return nil, fmt.Errorf("workflow %s/%s/%s not found", ns, workflowID, runID)
	}
	if err != nil {
		return nil, err
	}
	wf.HistoryLen++
	ev := &HistoryEvent{
		EventId:    wf.HistoryLen,
		EventTime:  nowRFC3339(),
		EventType:  eventType,
		Attributes: attrs,
	}
	key := fmt.Sprintf("wfh/%s/%s/%s/%020d", ns, wf.Execution.WorkflowId, wf.Execution.RunId, ev.EventId)
	if err := e.store.put(key, ev); err != nil {
		return nil, err
	}
	if err := e.store.put(fmt.Sprintf("wf/%s/%s/%s", ns, wf.Execution.WorkflowId, wf.Execution.RunId), wf); err != nil {
		return nil, err
	}
	return ev, nil
}

// GetWorkflowHistory returns events in [afterEventID+1, +pageSize] in
// the requested order. afterEventID==0 starts from the beginning.
// pageSize<=0 defaults to 100.
func (e *engine) GetWorkflowHistory(ns, workflowID, runID string, afterEventID int64, pageSize int, reverse bool) ([]HistoryEvent, int64, error) {
	if pageSize <= 0 {
		pageSize = 100
	}
	wf, ok, err := e.DescribeWorkflow(ns, workflowID, runID)
	if !ok {
		return nil, 0, fmt.Errorf("workflow not found")
	}
	if err != nil {
		return nil, 0, err
	}
	all, err := listInto[HistoryEvent](e.store, fmt.Sprintf("wfh/%s/%s/%s/", ns, wf.Execution.WorkflowId, wf.Execution.RunId))
	if err != nil {
		return nil, 0, err
	}
	if reverse {
		// Reverse in place.
		for i, j := 0, len(all)-1; i < j; i, j = i+1, j-1 {
			all[i], all[j] = all[j], all[i]
		}
	}
	out := make([]HistoryEvent, 0, pageSize)
	for _, ev := range all {
		if !reverse && ev.EventId <= afterEventID {
			continue
		}
		if reverse && afterEventID > 0 && ev.EventId >= afterEventID {
			continue
		}
		out = append(out, ev)
		if len(out) >= pageSize {
			break
		}
	}
	var nextCursor int64
	if len(out) == pageSize {
		nextCursor = out[len(out)-1].EventId
	}
	return out, nextCursor, nil
}

// QueryWorkflow dispatches the query through the dispatcher push path:
// if a worker is subscribed to the workflow's task queue, the query is
// pushed via OpcodeDeliverQuery and the call blocks for the worker's
// OpcodeRespondQuery (default 5s timeout, ctx-overrideable). Built-in
// synthetic queries (__stack_trace, __workflow_metadata) are handled
// engine-side without round-tripping a worker. ErrNoWorkersSubscribed
// is returned when the workflow has a task queue with no subscribers.
func (e *engine) QueryWorkflow(ns, workflowID, runID, queryType string, args any) (any, error) {
	return e.QueryWorkflowCtx(context.Background(), ns, workflowID, runID, queryType, args)
}

func (e *engine) QueryWorkflowCtx(ctx context.Context, ns, workflowID, runID, queryType string, args any) (any, error) {
	wf, ok, err := e.DescribeWorkflow(ns, workflowID, runID)
	if !ok {
		return nil, fmt.Errorf("workflow not found")
	}
	if err != nil {
		return nil, err
	}
	switch queryType {
	case "__stack_trace":
		return map[string]any{"stack": ""}, nil
	case "__workflow_metadata":
		return map[string]any{
			"workflowType": wf.Type.Name,
			"status":       wf.Status,
			"historyLen":   wf.HistoryLen,
		}, nil
	}
	if e.disp == nil {
		return nil, ErrNoWorkersSubscribed
	}
	var argBytes []byte
	if args != nil {
		argBytes, _ = json.Marshal(args)
	}
	token, ch, err := e.disp.PushQuery(ns, wf.TaskQueue, wf.Execution.WorkflowId, wf.Execution.RunId, queryType, argBytes)
	if err != nil {
		return nil, err
	}
	timeout := 5 * time.Second
	if dl, ok := ctx.Deadline(); ok {
		if d := time.Until(dl); d > 0 && d < timeout {
			timeout = d
		}
	}
	select {
	case res := <-ch:
		if res.errMsg != "" {
			return nil, fmt.Errorf("query failed: %s", res.errMsg)
		}
		var out any
		if len(res.result) > 0 {
			if err := json.Unmarshal(res.result, &out); err != nil {
				return string(res.result), nil
			}
		}
		return out, nil
	case <-ctx.Done():
		e.disp.CancelQuery(token)
		return nil, ctx.Err()
	case <-time.After(timeout):
		e.disp.CancelQuery(token)
		return nil, fmt.Errorf("query timeout after %s", timeout)
	}
}

// ResetWorkflow forks a workflow's history at eventID, writes a fresh
// run starting from a synthetic START event and the truncated history,
// and links the runs via wf.ResetPoints (stored as a memo key).
func (e *engine) ResetWorkflow(ns, workflowID, runID string, eventID int64, reason, identity string) (*WorkflowExecution, error) {
	src, ok, err := e.DescribeWorkflow(ns, workflowID, runID)
	if !ok {
		return nil, fmt.Errorf("workflow not found")
	}
	if err != nil {
		return nil, err
	}
	if eventID < 1 || eventID > src.HistoryLen {
		return nil, fmt.Errorf("invalid eventId %d (history has %d events)", eventID, src.HistoryLen)
	}
	all, err := listInto[HistoryEvent](e.store, fmt.Sprintf("wfh/%s/%s/%s/", ns, src.Execution.WorkflowId, src.Execution.RunId))
	if err != nil {
		return nil, err
	}
	newRun := newRunId()
	newWf := WorkflowExecution{
		Execution: ExecutionRef{WorkflowId: src.Execution.WorkflowId, RunId: newRun},
		Type:      src.Type,
		StartTime: nowRFC3339(),
		Status:    "WORKFLOW_EXECUTION_STATUS_RUNNING",
		TaskQueue: src.TaskQueue,
		Input:     src.Input,
		Memo: map[string]any{
			"resetFromRunId":   src.Execution.RunId,
			"resetFromEventId": eventID,
			"resetReason":      reason,
			"resetIdentity":    identity,
		},
	}
	// Persist the new execution shell first so appendHistory finds it.
	if err := e.store.put(fmt.Sprintf("wf/%s/%s/%s", ns, newWf.Execution.WorkflowId, newWf.Execution.RunId), newWf); err != nil {
		return nil, err
	}
	// Replay events [1..eventID] under the new runID with the same
	// EventId numbering so the fork is identical up to the cut.
	for _, ev := range all {
		if ev.EventId > eventID {
			break
		}
		key := fmt.Sprintf("wfh/%s/%s/%s/%020d", ns, newWf.Execution.WorkflowId, newWf.Execution.RunId, ev.EventId)
		if err := e.store.put(key, ev); err != nil {
			return nil, err
		}
	}
	newWf.HistoryLen = eventID
	if err := e.store.put(fmt.Sprintf("wf/%s/%s/%s", ns, newWf.Execution.WorkflowId, newWf.Execution.RunId), newWf); err != nil {
		return nil, err
	}
	if _, err := e.appendHistory(ns, newWf.Execution.WorkflowId, newWf.Execution.RunId, "WORKFLOW_EXECUTION_RESET", map[string]any{
		"baseRunId": src.Execution.RunId,
		"eventId":   eventID,
		"reason":    reason,
		"identity":  identity,
	}); err != nil {
		return nil, err
	}
	// Mark the source as terminated-by-reset; mirror Temporal semantics.
	src.Status = "WORKFLOW_EXECUTION_STATUS_TERMINATED"
	src.CloseTime = nowRFC3339()
	_ = e.store.put(fmt.Sprintf("wf/%s/%s/%s", ns, src.Execution.WorkflowId, src.Execution.RunId), src)
	// Deliver the forked run's first workflow task carrying its (truncated
	// + RESET) history so the worker replays from the fork point.
	if err := e.scheduleWorkflowTask(ns, newWf.Execution.WorkflowId, newWf.Execution.RunId); err != nil {
		return nil, err
	}
	out, _, _ := e.DescribeWorkflow(ns, newWf.Execution.WorkflowId, newWf.Execution.RunId)
	e.emit(Event{
		Kind:       "workflow.reset",
		Namespace:  ns,
		WorkflowID: out.Execution.WorkflowId,
		RunID:      out.Execution.RunId,
		Data:       out,
	})
	return out, nil
}

// ListWorkflowExecutions filters ListWorkflows by the supplied
// visibility query. The grammar:
//
//	expr   := orExpr
//	orExpr := andExpr ('OR' andExpr)*
//	andExpr:= notExpr ('AND' notExpr)*
//	notExpr:= 'NOT' notExpr | atom
//	atom   := '(' expr ')' | comparison
//	comparison := ident op value | ident 'IN' '(' values ')'
//	            | ident 'NOT' 'IN' '(' values ')'
//	            | ident 'BETWEEN' value 'AND' value
//	op     := '=' | '!=' | '>' | '<' | '>=' | '<='
//
// Empty query → no predicates → all rows match.
func (e *engine) ListWorkflowExecutions(ns, query string) ([]WorkflowExecution, error) {
	rows, err := e.ListWorkflows(ns)
	if err != nil {
		return nil, err
	}
	expr, err := parseVisibilityQuery(query)
	if err != nil {
		return nil, err
	}
	out := make([]WorkflowExecution, 0, len(rows))
	for i := range rows {
		if expr == nil || expr.eval(&rows[i]) {
			out = append(out, rows[i])
		}
	}
	return out, nil
}

// visNode is the evaluator interface; concrete types implement eval.
type visNode interface {
	eval(*WorkflowExecution) bool
}

type visAnd struct{ a, b visNode }

func (n visAnd) eval(wf *WorkflowExecution) bool { return n.a.eval(wf) && n.b.eval(wf) }

type visOr struct{ a, b visNode }

func (n visOr) eval(wf *WorkflowExecution) bool { return n.a.eval(wf) || n.b.eval(wf) }

type visNot struct{ a visNode }

func (n visNot) eval(wf *WorkflowExecution) bool { return !n.a.eval(wf) }

type visCmp struct {
	field string
	op    string // = != > < >= <= IN NOT_IN BETWEEN
	value string
	list  []string // for IN / NOT_IN
	low   string   // for BETWEEN
	high  string   // for BETWEEN
}

func (n visCmp) eval(wf *WorkflowExecution) bool { return matchCmp(wf, n) }

// ── tokenizer ───────────────────────────────────────────────────────

type vtok struct {
	kind string // ident | str | num | sym | kw
	val  string
}

func vtokenize(q string) ([]vtok, error) {
	var out []vtok
	i := 0
	for i < len(q) {
		c := q[i]
		switch {
		case c == ' ' || c == '\t' || c == '\n':
			i++
		case c == '(' || c == ')' || c == ',':
			out = append(out, vtok{"sym", string(c)})
			i++
		case c == '=':
			out = append(out, vtok{"sym", "="})
			i++
		case c == '!':
			if i+1 < len(q) && q[i+1] == '=' {
				out = append(out, vtok{"sym", "!="})
				i += 2
			} else {
				return nil, fmt.Errorf("unexpected '!' at %d", i)
			}
		case c == '>':
			if i+1 < len(q) && q[i+1] == '=' {
				out = append(out, vtok{"sym", ">="})
				i += 2
			} else {
				out = append(out, vtok{"sym", ">"})
				i++
			}
		case c == '<':
			if i+1 < len(q) && q[i+1] == '=' {
				out = append(out, vtok{"sym", "<="})
				i += 2
			} else {
				out = append(out, vtok{"sym", "<"})
				i++
			}
		case c == '\'' || c == '"':
			quote := c
			i++
			start := i
			for i < len(q) && q[i] != quote {
				i++
			}
			if i >= len(q) {
				return nil, fmt.Errorf("unterminated string starting at %d", start-1)
			}
			out = append(out, vtok{"str", q[start:i]})
			i++
		case (c >= '0' && c <= '9') || c == '-' || c == '+':
			start := i
			i++
			for i < len(q) && ((q[i] >= '0' && q[i] <= '9') || q[i] == '.') {
				i++
			}
			out = append(out, vtok{"num", q[start:i]})
		case (c >= 'A' && c <= 'Z') || (c >= 'a' && c <= 'z') || c == '_':
			start := i
			for i < len(q) && ((q[i] >= 'A' && q[i] <= 'Z') || (q[i] >= 'a' && q[i] <= 'z') || (q[i] >= '0' && q[i] <= '9') || q[i] == '_') {
				i++
			}
			word := q[start:i]
			upper := strings.ToUpper(word)
			switch upper {
			case "AND", "OR", "NOT", "IN", "BETWEEN":
				out = append(out, vtok{"kw", upper})
			default:
				out = append(out, vtok{"ident", word})
			}
		default:
			return nil, fmt.Errorf("unexpected char %q at %d", c, i)
		}
	}
	return out, nil
}

// ── parser ─────────────────────────────────────────────────────────

type vparser struct {
	toks []vtok
	pos  int
}

func (p *vparser) peek() *vtok {
	if p.pos >= len(p.toks) {
		return nil
	}
	return &p.toks[p.pos]
}

func (p *vparser) eat() *vtok {
	t := p.peek()
	if t != nil {
		p.pos++
	}
	return t
}

func (p *vparser) expectSym(s string) error {
	t := p.eat()
	if t == nil || t.kind != "sym" || t.val != s {
		return fmt.Errorf("expected %q", s)
	}
	return nil
}

func parseVisibilityQuery(q string) (visNode, error) {
	q = strings.TrimSpace(q)
	if q == "" {
		return nil, nil
	}
	toks, err := vtokenize(q)
	if err != nil {
		return nil, err
	}
	p := &vparser{toks: toks}
	n, err := p.parseOr()
	if err != nil {
		return nil, err
	}
	if p.pos != len(toks) {
		return nil, fmt.Errorf("trailing tokens at pos %d", p.pos)
	}
	return n, nil
}

func (p *vparser) parseOr() (visNode, error) {
	a, err := p.parseAnd()
	if err != nil {
		return nil, err
	}
	for {
		t := p.peek()
		if t == nil || t.kind != "kw" || t.val != "OR" {
			return a, nil
		}
		p.eat()
		b, err := p.parseAnd()
		if err != nil {
			return nil, err
		}
		a = visOr{a, b}
	}
}

func (p *vparser) parseAnd() (visNode, error) {
	a, err := p.parseNot()
	if err != nil {
		return nil, err
	}
	for {
		t := p.peek()
		if t == nil || t.kind != "kw" || t.val != "AND" {
			return a, nil
		}
		p.eat()
		b, err := p.parseNot()
		if err != nil {
			return nil, err
		}
		a = visAnd{a, b}
	}
}

func (p *vparser) parseNot() (visNode, error) {
	t := p.peek()
	if t != nil && t.kind == "kw" && t.val == "NOT" {
		// "NOT IN" only valid as part of an IN comparison; we look ahead
		// for an ident before consuming.
		if p.pos+1 < len(p.toks) && p.toks[p.pos+1].kind == "ident" {
			// Could still be "NOT atom"; only treat as NOT IN if the
			// ident is followed by IN — but that's parsed inside the
			// atom's comparison. So here, NOT followed by ident is a
			// NOT-of-comparison.
		}
		p.eat()
		inner, err := p.parseNot()
		if err != nil {
			return nil, err
		}
		return visNot{inner}, nil
	}
	return p.parseAtom()
}

func (p *vparser) parseAtom() (visNode, error) {
	t := p.peek()
	if t == nil {
		return nil, fmt.Errorf("unexpected end of query")
	}
	if t.kind == "sym" && t.val == "(" {
		p.eat()
		n, err := p.parseOr()
		if err != nil {
			return nil, err
		}
		if err := p.expectSym(")"); err != nil {
			return nil, err
		}
		return n, nil
	}
	return p.parseComparison()
}

func (p *vparser) parseComparison() (visNode, error) {
	id := p.eat()
	if id == nil || id.kind != "ident" {
		return nil, fmt.Errorf("expected identifier, got %+v", id)
	}
	op := p.eat()
	if op == nil {
		return nil, fmt.Errorf("expected operator after %q", id.val)
	}
	// IN / NOT IN / BETWEEN.
	if op.kind == "kw" {
		switch op.val {
		case "IN":
			vs, err := p.parseValueList()
			if err != nil {
				return nil, err
			}
			return visCmp{field: id.val, op: "IN", list: vs}, nil
		case "NOT":
			n2 := p.eat()
			if n2 == nil || n2.kind != "kw" || n2.val != "IN" {
				return nil, fmt.Errorf("expected IN after NOT")
			}
			vs, err := p.parseValueList()
			if err != nil {
				return nil, err
			}
			return visCmp{field: id.val, op: "NOT_IN", list: vs}, nil
		case "BETWEEN":
			lo, err := p.parseValue()
			if err != nil {
				return nil, err
			}
			and := p.eat()
			if and == nil || and.kind != "kw" || and.val != "AND" {
				return nil, fmt.Errorf("expected AND in BETWEEN")
			}
			hi, err := p.parseValue()
			if err != nil {
				return nil, err
			}
			return visCmp{field: id.val, op: "BETWEEN", low: lo, high: hi}, nil
		}
		return nil, fmt.Errorf("unexpected keyword %q", op.val)
	}
	if op.kind != "sym" {
		return nil, fmt.Errorf("expected operator, got %+v", op)
	}
	switch op.val {
	case "=", "!=", ">", "<", ">=", "<=":
		v, err := p.parseValue()
		if err != nil {
			return nil, err
		}
		return visCmp{field: id.val, op: op.val, value: v}, nil
	}
	return nil, fmt.Errorf("unexpected operator %q", op.val)
}

func (p *vparser) parseValue() (string, error) {
	t := p.eat()
	if t == nil {
		return "", fmt.Errorf("expected value")
	}
	switch t.kind {
	case "str", "num", "ident":
		return t.val, nil
	}
	return "", fmt.Errorf("expected value, got %+v", t)
}

func (p *vparser) parseValueList() ([]string, error) {
	if err := p.expectSym("("); err != nil {
		return nil, err
	}
	var out []string
	for {
		v, err := p.parseValue()
		if err != nil {
			return nil, err
		}
		out = append(out, v)
		t := p.peek()
		if t != nil && t.kind == "sym" && t.val == "," {
			p.eat()
			continue
		}
		break
	}
	if err := p.expectSym(")"); err != nil {
		return nil, err
	}
	return out, nil
}

// ── evaluator ───────────────────────────────────────────────────────

func matchCmp(wf *WorkflowExecution, c visCmp) bool {
	got := fieldValue(wf, c.field)
	if c.field == "ExecutionStatus" {
		got = normalizeStatus(got)
	}
	switch c.op {
	case "=":
		return got == compareValue(c.field, c.value)
	case "!=":
		return got != compareValue(c.field, c.value)
	case ">":
		return got > compareValue(c.field, c.value)
	case "<":
		return got < compareValue(c.field, c.value)
	case ">=":
		return got >= compareValue(c.field, c.value)
	case "<=":
		return got <= compareValue(c.field, c.value)
	case "IN":
		for _, v := range c.list {
			if got == compareValue(c.field, v) {
				return true
			}
		}
		return false
	case "NOT_IN":
		for _, v := range c.list {
			if got == compareValue(c.field, v) {
				return false
			}
		}
		return true
	case "BETWEEN":
		lo := compareValue(c.field, c.low)
		hi := compareValue(c.field, c.high)
		return got >= lo && got <= hi
	}
	return false
}

func fieldValue(wf *WorkflowExecution, f string) string {
	switch f {
	case "WorkflowType":
		return wf.Type.Name
	case "WorkflowId":
		return wf.Execution.WorkflowId
	case "RunId":
		return wf.Execution.RunId
	case "ExecutionStatus":
		return wf.Status
	case "StartTime":
		return wf.StartTime
	case "CloseTime":
		return wf.CloseTime
	case "TaskQueue":
		return wf.TaskQueue
	}
	if v, ok := wf.SearchAttrs[f]; ok {
		return fmt.Sprint(v)
	}
	return ""
}

func compareValue(field, want string) string {
	if field == "ExecutionStatus" {
		return normalizeStatus(want)
	}
	return want
}

func normalizeStatus(s string) string {
	s = strings.ToUpper(s)
	if !strings.HasPrefix(s, "WORKFLOW_EXECUTION_STATUS_") {
		s = "WORKFLOW_EXECUTION_STATUS_" + s
	}
	return s
}

// ── schedules ───────────────────────────────────────────────────────

func (e *engine) CreateSchedule(s Schedule) error {
	if s.ScheduleId == "" {
		return fmt.Errorf("scheduleId required")
	}
	if s.Namespace == "" {
		return fmt.Errorf("namespace required")
	}
	s.Info.CreateTime = nowRFC3339()
	if next, ok := nextScheduleFire(s, time.Now().UTC()); ok {
		s.Info.NextActionTime = next.UTC().Format(time.RFC3339)
	}
	if err := e.store.put(fmt.Sprintf("sc/%s/%s", s.Namespace, s.ScheduleId), s); err != nil {
		return err
	}
	e.emit(Event{Kind: "schedule.created", Namespace: s.Namespace, ScheduleID: s.ScheduleId, Data: s})
	return nil
}

func (e *engine) DescribeSchedule(ns, id string) (*Schedule, bool, error) {
	var s Schedule
	ok, err := e.store.get(fmt.Sprintf("sc/%s/%s", ns, id), &s)
	if !ok {
		return nil, false, err
	}
	return &s, true, err
}

func (e *engine) ListSchedules(ns string) ([]Schedule, error) {
	return listInto[Schedule](e.store, fmt.Sprintf("sc/%s/", ns))
}

func (e *engine) DeleteSchedule(ns, id string) error {
	// Drop the failure throttle with the row it belongs to. A schedule
	// deleted while broken never fires again, so nothing else would ever
	// clear its counter.
	e.fails.ok(scheduleKey(e.principal, ns, id))
	return e.store.del(fmt.Sprintf("sc/%s/%s", ns, id))
}

func (e *engine) PauseSchedule(ns, id string, paused bool, note string) error {
	s, ok, err := e.DescribeSchedule(ns, id)
	if !ok || err != nil {
		return fmt.Errorf("schedule not found")
	}
	s.State.Paused = paused
	s.State.Note = note
	s.Info.UpdateTime = nowRFC3339()
	return e.store.put(fmt.Sprintf("sc/%s/%s", ns, id), s)
}

// ── search attributes ───────────────────────────────────────────────

var validSearchAttrTypes = map[string]bool{
	"Keyword": true, "Text": true, "Int": true, "Double": true,
	"Bool": true, "Datetime": true, "KeywordList": true,
}

// reservedSearchAttrNames are the system fields the visibility evaluator
// interprets directly. Allowing user code to register a custom attribute
// with one of these names would let it shadow the real workflow field
// (e.g. ExecutionStatus="*") and silently widen list results. Names
// starting with "Temporal" are reserved for upstream-compat fields.
var reservedSearchAttrNames = map[string]bool{
	"WorkflowType": true, "WorkflowId": true, "RunId": true,
	"ExecutionStatus": true, "StartTime": true, "CloseTime": true,
	"TaskQueue": true,
}

func (e *engine) AddSearchAttribute(ns string, sa SearchAttribute) error {
	if _, err := e.namespace(ns); err != nil {
		return err
	}
	if sa.Name == "" {
		return fmt.Errorf("search attribute name required")
	}
	if reservedSearchAttrNames[sa.Name] || strings.HasPrefix(sa.Name, "Temporal") {
		return fmt.Errorf("search attribute %q is reserved", sa.Name)
	}
	if !validSearchAttrTypes[sa.Type] {
		return fmt.Errorf("invalid search attribute type %q", sa.Type)
	}
	key := fmt.Sprintf("sa/%s/%s", ns, sa.Name)
	var existing SearchAttribute
	if ok, _ := e.store.get(key, &existing); ok {
		return fmt.Errorf("search attribute %q already exists", sa.Name)
	}
	return e.store.put(key, sa)
}

func (e *engine) ListSearchAttributes(ns string) ([]SearchAttribute, error) {
	return listInto[SearchAttribute](e.store, fmt.Sprintf("sa/%s/", ns))
}

// ── namespace metadata ──────────────────────────────────────────────

// NamespaceMetadataPatch carries the fields a metadata update may set.
// Empty string fields are ignored; non-nil maps fully replace.
type NamespaceMetadataPatch struct {
	Description             *string           `json:"description,omitempty"`
	OwnerEmail              *string           `json:"ownerEmail,omitempty"`
	Retention               *string           `json:"retention,omitempty"`
	HistoryArchivalState    *string           `json:"historyArchivalState,omitempty"`
	HistoryArchivalUri      *string           `json:"historyArchivalUri,omitempty"`
	VisibilityArchivalState *string           `json:"visibilityArchivalState,omitempty"`
	VisibilityArchivalUri   *string           `json:"visibilityArchivalUri,omitempty"`
	CustomData              map[string]string `json:"customData,omitempty"`
}

func (e *engine) UpdateNamespaceMetadata(name string, p NamespaceMetadataPatch) (*Namespace, error) {
	n, ok, err := e.DescribeNamespace(name)
	if !ok {
		return nil, fmt.Errorf("namespace %q not registered", name)
	}
	if err != nil {
		return nil, err
	}
	if p.Description != nil {
		n.NamespaceInfo.Description = *p.Description
	}
	if p.OwnerEmail != nil {
		n.NamespaceInfo.OwnerEmail = *p.OwnerEmail
	}
	if p.Retention != nil {
		n.Config.WorkflowExecutionRetentionTtl = *p.Retention
	}
	if p.HistoryArchivalState != nil {
		n.Config.HistoryArchivalState = *p.HistoryArchivalState
	}
	if p.HistoryArchivalUri != nil {
		n.Config.HistoryArchivalUri = *p.HistoryArchivalUri
	}
	if p.VisibilityArchivalState != nil {
		n.Config.VisibilityArchivalState = *p.VisibilityArchivalState
	}
	if p.VisibilityArchivalUri != nil {
		n.Config.VisibilityArchivalUri = *p.VisibilityArchivalUri
	}
	if p.CustomData != nil {
		n.Config.CustomData = p.CustomData
	}
	if err := e.store.put("ns/"+name, *n); err != nil {
		return nil, err
	}
	return n, nil
}

// ── workflow user metadata ──────────────────────────────────────────

func (e *engine) UpdateWorkflowMetadata(ns, workflowID, runID string, m WorkflowUserMetadata) (*WorkflowExecution, error) {
	wf, ok, err := e.DescribeWorkflow(ns, workflowID, runID)
	if !ok {
		return nil, fmt.Errorf("workflow not found")
	}
	if err != nil {
		return nil, err
	}
	prev := wf.UserMetadata
	cur := WorkflowUserMetadata{}
	if prev != nil {
		cur = *prev
	}
	if m.Summary != "" {
		cur.Summary = m.Summary
	}
	if m.Details != "" {
		cur.Details = m.Details
	}
	cur.UpdatedBy = m.UpdatedBy
	cur.UpdatedAt = nowRFC3339()
	wf.UserMetadata = &cur
	if err := e.store.put(fmt.Sprintf("wf/%s/%s/%s", ns, wf.Execution.WorkflowId, wf.Execution.RunId), wf); err != nil {
		return nil, err
	}
	return wf, nil
}

// ── batches ─────────────────────────────────────────────────────────

func (e *engine) StartBatch(b BatchOperation) (*BatchOperation, error) {
	if b.BatchId == "" {
		b.BatchId = "batch-" + newRunId()[:12]
	}
	b.State = "BATCH_OPERATION_STATE_RUNNING"
	b.StartTime = nowRFC3339()
	if err := e.store.put(fmt.Sprintf("bt/%s/%s", b.Namespace, b.BatchId), b); err != nil {
		return nil, err
	}
	return &b, nil
}

func (e *engine) DescribeBatch(ns, id string) (*BatchOperation, bool, error) {
	var b BatchOperation
	ok, err := e.store.get(fmt.Sprintf("bt/%s/%s", ns, id), &b)
	if !ok {
		return nil, false, err
	}
	return &b, true, err
}

func (e *engine) ListBatches(ns string) ([]BatchOperation, error) {
	return listInto[BatchOperation](e.store, fmt.Sprintf("bt/%s/", ns))
}

// TerminateBatch marks a batch as terminated and appends a terminate
// event to the namespace's batch event log. Idempotent: if the batch is
// already terminated, returns the existing record without re-emitting.
func (e *engine) TerminateBatch(ns, batchID, reason, identity string) (*BatchOperation, error) {
	b, ok, err := e.DescribeBatch(ns, batchID)
	if !ok {
		return nil, fmt.Errorf("batch not found")
	}
	if err != nil {
		return nil, err
	}
	if b.State == "BATCH_OPERATION_STATE_TERMINATED" {
		return b, nil
	}
	b.State = "BATCH_OPERATION_STATE_TERMINATED"
	b.CloseTime = nowRFC3339()
	if err := e.store.put(fmt.Sprintf("bt/%s/%s", ns, b.BatchId), b); err != nil {
		return nil, err
	}
	e.emit(Event{
		Kind:      "batch.terminated",
		Namespace: ns,
		BatchID:   b.BatchId,
		Data: map[string]any{
			"reason":   reason,
			"identity": identity,
		},
	})
	return b, nil
}

// ── deployments ─────────────────────────────────────────────────────

// CreateDeployment registers a new deployment series. Returns 409-style
// error if the name already exists in the namespace. Caller must check
// for "already exists" substring to map to HTTP 409.
func (e *engine) CreateDeployment(ns, name, description, ownerEmail, defaultCompute string) (*Deployment, error) {
	if name == "" {
		return nil, fmt.Errorf("name required")
	}
	key := fmt.Sprintf("dp/%s/%s", ns, name)
	var existing Deployment
	if ok, _ := e.store.get(key, &existing); ok {
		return nil, fmt.Errorf("deployment %q already exists", name)
	}
	d := Deployment{
		Name:           name,
		Namespace:      ns,
		Description:    description,
		OwnerEmail:     ownerEmail,
		DefaultCompute: defaultCompute,
		Versions:       []DeploymentVersion{},
		CreateTime:     nowRFC3339(),
	}
	if err := e.store.put(key, d); err != nil {
		return nil, err
	}
	return &d, nil
}

func (e *engine) ListDeployments(ns string) ([]Deployment, error) {
	return listInto[Deployment](e.store, fmt.Sprintf("dp/%s/", ns))
}

// DescribeDeployment returns a single deployment by name.
func (e *engine) DescribeDeployment(ns, name string) (*Deployment, bool, error) {
	var d Deployment
	ok, err := e.store.get(fmt.Sprintf("dp/%s/%s", ns, name), &d)
	if !ok {
		return nil, false, err
	}
	return &d, true, err
}

// ── nexus ───────────────────────────────────────────────────────────

func (e *engine) CreateNexusEndpoint(n NexusEndpoint) error {
	if n.Name == "" {
		return fmt.Errorf("nexus endpoint name required")
	}
	n.CreateTime = nowRFC3339()
	return e.store.put(fmt.Sprintf("nx/%s/%s", n.Namespace, n.Name), n)
}

func (e *engine) ListNexusEndpoints(ns string) ([]NexusEndpoint, error) {
	return listInto[NexusEndpoint](e.store, fmt.Sprintf("nx/%s/", ns))
}

// ListAllNexusEndpoints aggregates Nexus endpoints across all namespaces.
// Used by the cross-namespace top-level UI route.
func (e *engine) ListAllNexusEndpoints() ([]NexusEndpoint, error) {
	return listInto[NexusEndpoint](e.store, "nx/")
}

// ── identities ──────────────────────────────────────────────────────

func (e *engine) GrantIdentity(i Identity) error {
	if i.Email == "" || i.Namespace == "" {
		return fmt.Errorf("email + namespace required")
	}
	i.GrantTime = nowRFC3339()
	return e.store.put(fmt.Sprintf("id/%s/%s", i.Namespace, i.Email), i)
}

func (e *engine) ListIdentities(ns string) ([]Identity, error) {
	return listInto[Identity](e.store, fmt.Sprintf("id/%s/", ns))
}

// ── cron sweeper ────────────────────────────────────────────────────

// scheduleKey identifies a schedule across every tenant's shard. Also the
// tracker key, so counts follow the schedule and not the engine view.
func scheduleKey(p Principal, ns, id string) string { return p.String() + "|" + ns + "|" + id }

// sweepAllKey tracks whole-sweep failures under the same throttle as a
// single broken schedule. CreateSchedule rejects an empty namespace and an
// empty scheduleId, so this all-empty triple can never collide with a real
// schedule's key.
var sweepAllKey = scheduleKey(Principal{}, "", "")

// The cron sweeper's report messages. Consts rather than literals so a test
// asserting on them cannot drift from the line the engine actually writes.
const (
	msgActionFailed    = "tasks: cron schedule action failed"
	msgActionRecovered = "tasks: cron schedule action recovered"
	msgSweepFailed     = "tasks: schedule sweep failed, no schedule fired this tick"
)

// unreached names the fault in a fire that STARTED but has nowhere to go:
// the workflow task is enqueued on (ns, queue) and sits in the dispatcher's
// backlog until some worker subscribes. nil when a live subscriber is there
// to take it.
//
// This is what a cron entry naming a namespace nobody serves looks like now
// that a namespace is created on demand: the start succeeds, the run is
// recorded, and nothing ever executes it. It is also what a schedule whose
// worker died looks like, and a schedule whose task queue was renamed — the
// three shapes of "the nightly backup did not run" that no start-time check
// can see. Ask AFTER the start, because the enqueue prunes subscriptions
// whose peer the transport can no longer reach (deliverWFLocked): a worker
// that died without unsubscribing is gone from the registry by the time we
// look.
func (e *engine) unreached(ns, queue string) error {
	if e.disp != nil && e.disp.HasSubscribers(ns, queue, kindWorkflow) {
		return nil
	}
	return fmt.Errorf("%w %q in namespace %q", ErrNoWorkersSubscribed, queue, ns)
}

// reportFire is the ONE place a schedule fire's outcome is reported. A nil
// fault closes any open streak; a non-nil one opens or extends it, throttled.
//
// Isolation must not mean SILENCE. cloud's clients/cron fires a JobWorkflow
// per entry; when its activity could not read the entry's ConfigMap (the
// cloud ServiceAccount was missing the RBAC) every fire failed for 11 days
// while the engine reported nothing and every dashboard read healthy — six
// nightly backups produced nothing and it was caught only by hand-reading
// the engine's durable records. So every field needed to act on it goes on
// the line: which org, which namespace, which schedule, what it was trying
// to start and where, and how long it has been failing.
//
// Throttled because a broken schedule stays broken: at a 5s tick that is
// ~17k identical lines a day, which buries the signal as thoroughly as
// silence did. At failReportThrottle=60 the heartbeat is one line per ≈5
// minutes, so any 15-minute dashboard window still holds at least three.
func (e *engine) reportFire(p Principal, s Schedule, fault error) {
	key := scheduleKey(p, s.Namespace, s.ScheduleId)
	if fault == nil {
		if n := e.fails.ok(key); n > 0 {
			// Closes the loop the throttle opens: without a recovery line an
			// operator cannot tell a schedule that was fixed from one whose
			// next throttled error line simply has not come round yet.
			e.log.Info(msgActionRecovered,
				"org", p.Org,
				"namespace", s.Namespace,
				"scheduleId", s.ScheduleId,
				"afterConsecutiveFailures", n)
		}
		return
	}
	if n, report := e.fails.fail(key); report {
		e.log.Error(msgActionFailed,
			"org", p.Org,
			"namespace", s.Namespace,
			"scheduleId", s.ScheduleId,
			"workflowType", s.Action.WorkflowType.Name,
			"taskQueue", s.Action.TaskQueue,
			"consecutiveFailures", n,
			"error", fault)
	}
}

// runScheduler ticks every 5s and fires due schedules through StartWorkflow.
// Cron specs are parsed via robfig/cron with seconds-optional 5/6-field
// syntax; first invalid spec is skipped without aborting the sweep.
func (e *engine) runScheduler(stop <-chan struct{}) {
	t := time.NewTicker(5 * time.Second)
	defer t.Stop()
	for {
		select {
		case <-stop:
			return
		case <-t.C:
			e.sweepOnce()
		}
	}
}

// sweepOnce runs one scheduler tick. Split out of runScheduler's select so
// tests can drive a tick without waiting on the 5s ticker.
//
// The sweep error used to be dropped on the floor (`_ = e.sweepSchedules()`).
// That made a TOTAL sweep failure — the store refusing the cross-org scan,
// say — indistinguishable from a quiet night: cron stops firing for every
// org at once and nothing anywhere says so.
func (e *engine) sweepOnce() {
	if err := e.sweepSchedules(); err != nil {
		if n, report := e.fails.fail(sweepAllKey); report {
			e.log.Error(msgSweepFailed,
				"org", e.principal.Org, "consecutiveFailures", n, "error", err)
		}
	} else {
		e.fails.ok(sweepAllKey)
	}
	e.sweepCanceling()
	e.sweepLeases()
}

// sweepLeases reaps expired activity leases across every namespace the
// (unscoped) engine can see, so a claimed activity stranded by a worker that
// died and never came back is eventually reclaimed even with no fresh claim
// to trigger reapExpiredLeases on the hot path. In the org-scoped cloud
// deployment the per-org shards are invisible to this root sweep — there
// reclamation rides the claim path (a restarting/peer worker) instead.
func (e *engine) sweepLeases() {
	namespaces, err := e.ListNamespaces()
	if err != nil {
		return
	}
	for _, ns := range namespaces {
		name := ns.NamespaceInfo.Name
		unlock := e.runMu.lock(e.principal.String() + "|claim|" + name)
		_ = e.reapExpiredLeases(name)
		unlock()
	}
}

func (e *engine) sweepSchedules() error {
	now := time.Now().UTC()
	// Phase 1: COLLECT due schedules. Firing inside the List callback would
	// nest a query on the same single-connection SQLite shard and deadlock,
	// so the scan only gathers.
	type due struct {
		principal Principal
		s         Schedule
	}
	var dues []due
	if err := e.store.listEveryTenant("sc/", func(p Principal, _ string, body []byte) error {
		var s Schedule
		if err := unmarshal(body, &s); err != nil {
			return nil // skip one corrupt record, never poison the sweep
		}
		if s.State.Paused {
			return nil
		}
		next, ok := nextScheduleFire(s, anchorTime(s, now))
		if !ok || next.After(now) {
			return nil
		}
		dues = append(dues, due{principal: p, s: s})
		return nil
	}); err != nil {
		return err
	}
	// Phase 2: fire AT MOST ONE action per schedule per sweep and re-anchor
	// at NOW. A schedule that missed many ticks (engine downtime, a long-
	// paused window) must not storm its whole backlog on recovery — cron
	// semantics are "run on schedule", not "replay history". Sub-5s specs
	// quantize to the sweep cadence.
	for _, d := range dues {
		s := d.s
		oe := e.As(d.principal)
		// Stamp the originating schedule on the run. The 11-day incident's
		// fault was one layer below this call — StartWorkflow succeeded 4489
		// times while the ACTIVITY it started failed every time — and without
		// this attribute the failure report one layer down can only say "some
		// JobWorkflow is broken", never "the nightly backup has not run".
		wf, startErr := oe.startWorkflowFull(s.Namespace, "", "", s.Action.WorkflowType, s.Action.TaskQueue, s.Action.Input, "",
			map[string]any{searchAttrScheduleID: s.ScheduleId}, nil, "")
		fault := startErr
		if fault == nil {
			fault = e.unreached(s.Namespace, wf.TaskQueue)
		}
		e.reportFire(d.principal, s, fault)
		if startErr != nil {
			// Per-schedule isolation: a broken action skips this entry and
			// the sweep continues; it is re-attempted next sweep.
			//
			// A failed fire is deliberately NOT re-anchored, so the entry
			// stays due and retries on every 5s tick. A fire that STARTED is
			// re-anchored even when it went unreached, because the run
			// exists: re-firing it every tick would spawn a run every 5s on
			// the strength of a report.
			continue
		}
		s.Info.ActionCount++
		s.Info.UpdateTime = now.Format(time.RFC3339)
		if n2, ok2 := nextScheduleFire(s, now); ok2 {
			s.Info.NextActionTime = n2.UTC().Format(time.RFC3339)
		}
		if err := oe.store.put(fmt.Sprintf("sc/%s/%s", s.Namespace, s.ScheduleId), s); err != nil {
			return err
		}
	}
	return nil
}

// anchorTime returns the timestamp from which the next cron firing
// should be computed. Uses the last update time if present, otherwise
// the create time, otherwise now.
func anchorTime(s Schedule, fallback time.Time) time.Time {
	for _, ts := range []string{s.Info.UpdateTime, s.Info.CreateTime} {
		if ts == "" {
			continue
		}
		if t, err := time.Parse(time.RFC3339, ts); err == nil {
			return t
		}
	}
	return fallback
}

// cronParser parses both 5-field (minute precision) and 6-field
// (with-seconds) crontab syntax. Robfig defaults to 5 fields; we add
// the optional seconds descriptor so workflows can be sub-minute.
var cronParser = cron.NewParser(cron.SecondOptional | cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow | cron.Descriptor)

// nextScheduleFire computes the next fire time across all configured
// cron strings AND interval specs, returning the earliest. (false, _)
// means the schedule has no valid spec. An interval fires one interval
// after the anchor (the last fire / create time), so re-anchoring on
// each fire yields steady every-N cadence.
func nextScheduleFire(s Schedule, after time.Time) (time.Time, bool) {
	var earliest time.Time
	for _, cs := range s.Spec.CronString {
		sched, err := cronParser.Parse(cs)
		if err != nil {
			continue
		}
		t := sched.Next(after)
		if earliest.IsZero() || t.Before(earliest) {
			earliest = t
		}
	}
	for _, iv := range s.Spec.Interval {
		d, err := time.ParseDuration(iv.Interval)
		if err != nil || d <= 0 {
			continue
		}
		t := after.Add(d)
		if earliest.IsZero() || t.Before(earliest) {
			earliest = t
		}
	}
	if earliest.IsZero() {
		return time.Time{}, false
	}
	return earliest, true
}

func unmarshal(b []byte, v any) error {
	if err := jsonUnmarshal(b, v); err != nil {
		return fmt.Errorf("decode: %w", err)
	}
	return nil
}