perf: scale agent execution with typed execution state, a claim dispatcher, and owned continuation

Replace history-derived scheduling with an agent execution engine. Each agent
has a typed execution head of pointers, monotonic boundaries and summaries
recomputed at commit, plus immutable opening lineage on model contexts, config
changes and checkpoints. One pure selector decides next work for claim, owned
continuation, release and recovery. Semantic transitions run through a
cell-bound unit of work that plans locks up front in sorted order, supports
savepoints, and publishes each changed agent's head and queue in two pipelined
statements before enqueueing webhooks and post-commit notifications. Execution
writes live in a private query package enforced by SQL rules and a Go analyzer.

Workers run a bounded claim dispatcher, keep an owned agent across ready steps
within a configurable budget, and accept model output, select and prepare the
next call in one transaction.

Migration 052 builds every existing agent's execution state from history,
validates it, and drops the history frontier functions. It requires a
writer-stop maintenance window.

Also fixes:
- continuations stranded when a turn carried unanswered inputs from an earlier
  turn
- child-before-parent and machine/parent lock-order hazards
- subagent launch reusing a machine grant deleted while it waited
- project and organization deletion skipping event webhooks

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

Author: ksarangmath

.golangci.yml

@@ -141,8 +141,11 @@ linters:
   exclusions:
     generated: disable
     rules:
-      - path: internal/storage/internal/storeutil/pool\.go
-        source: '\.Query\(ctx, sql, args\.\.\.\)'
+      - path: internal/storage/internal/agentexecution/internal/executiondb/batch\.go
+        linters:
+          - intrange
+      - path: internal/storage/internal/(storeutil/pool|agentexecution/database)\.go
+        source: '(\.Query\(ctx, (sql|query), args\.\.\.\)|b\.BatchResults\.Query\(\))'
         linters:
           - rowserrcheck
       # oapi-codegen's oneOf wrappers intentionally mix value accessors with
@@ -180,7 +183,7 @@ linters:
       - linters:
           - lll
         source: '^\s*\)\.[A-Za-z]'
-      - path: (internal/storage/internal/dbsqlc/|internal/machinedaemon/statedb/internal/dbsqlc/)
+      - path: (internal/storage/internal/dbsqlc/|internal/storage/internal/agentexecution/internal/executiondb/|internal/machinedaemon/statedb/internal/dbsqlc/)
         linters:
           - errcheck
           - govet

Makefile

@@ -23,6 +23,7 @@ MIGRATION_CHECK ?= $(GO) -C tools/ci run ./migrationcheck
 OMNARAD_VERSION ?= 0.0.0-dev
 MIGRATION_DIRS := migrations internal/machinedaemon/statedb/migrations
 SQLC_OWNED_PATHS := internal/storage/internal/dbsqlc internal/storage/queries \
+	internal/storage/internal/agentexecution/internal/executiondb internal/storage/internal/agentexecution/queries \
 	internal/machinedaemon/statedb/internal/dbsqlc internal/machinedaemon/statedb/queries
 OMNARALINT_SOURCES := $(shell find tools/omnaralint -name '*.go' -print)
 GO_MODULE_DIRS := . observability tools/ci tools/goose tools/omnaralint
@@ -259,6 +260,7 @@ sqlc-check:
 
 sql-rules:
 	CGO_ENABLED=0 $(GO) -C tools/ci run ./sqlrules ../../internal/storage/queries
+	CGO_ENABLED=0 $(GO) -C tools/ci run ./sqlrules ../../internal/storage/internal/agentexecution/queries
 
 sqlc-vet:
 	$(SQLC) vet -f sqlc.yaml
@@ -267,7 +269,7 @@ migrate-test-db:
 	$(TEST_DB_ENV) $(GO) test -count=1 -run '^TestMigrateLocalDatabase$$' ./internal/testutil/integrationdb
 
 sqlc-vet-db:
-	SQLC_DATABASE_URL=$${SQLC_DATABASE_URL:-$${OMNARA_TEST_DATABASE_URL:-$(TEST_DATABASE_URL)}} $(SQLC) vet -f sqlc.vet-db.yaml
+	OMNARA_TEST_DATABASE_URL=$${SQLC_DATABASE_URL:-$${OMNARA_TEST_DATABASE_URL:-$(TEST_DATABASE_URL)}} $(GO) test -count=1 -tags=integration ./internal/storage/internal/agentexecution -run '^TestExecutionSQLCPrepare$$'
 
 sqlc-vet-local-db: db-up migrate-test-db sqlc-vet-db
 

cmd/worker/main.go

@@ -241,8 +241,13 @@ func main() {
 		StreamLog:       log,
 	})
 	kernelWorker := workerpkg.NewWorker(store.Execution(), executor, workerpkg.Options{
+		Continuation: &workerpkg.ContinuationOptions{
+			MaxModelStarts: cfg.WorkerContinuationMaxModelStarts,
+			MaxDuration:    cfg.WorkerContinuationMaxDuration,
+		},
 		Log:               log,
 		Capacity:          cfg.WorkerCapacity,
+		ClaimConcurrency:  cfg.WorkerClaimConcurrency,
 		AsyncToolCapacity: cfg.WorkerAsyncToolCapacity,
 		ControlSubscriber: redisBus,
 	})

compose.yaml

@@ -126,6 +126,7 @@ services:
       OMNARA_WORKER_EVENT_WEBHOOK_CONCURRENCY: ${OMNARA_WORKER_EVENT_WEBHOOK_CONCURRENCY:-128}
       OMNARA_EVENT_WEBHOOK_PER_ORG_CONCURRENCY: ${OMNARA_EVENT_WEBHOOK_PER_ORG_CONCURRENCY:-128}
       OMNARA_WORKER_CAPACITY: ${OMNARA_WORKER_CAPACITY:-4}
+      OMNARA_WORKER_CLAIM_CONCURRENCY: ${OMNARA_WORKER_CLAIM_CONCURRENCY:-4}
       OMNARA_WORKER_INBOX_CAPACITY: ${OMNARA_WORKER_INBOX_CAPACITY:-4}
       OMNARA_WORKER_DISCORD_CAPACITY: ${OMNARA_WORKER_DISCORD_CAPACITY:-1024}
       OMNARA_WORKER_ASYNC_TOOL_CAPACITY: ${OMNARA_WORKER_ASYNC_TOOL_CAPACITY:-32}

docs/self-hosting/configuration.mdx

@@ -28,11 +28,22 @@ All configuration is via environment variables. Use `.env.example` in the repo a
 | `OMNARA_WORKER_EVENT_WEBHOOK_CONCURRENCY` | Maximum event webhook deliveries in flight per worker process | `128` | No |
 | `OMNARA_EVENT_WEBHOOK_PER_ORG_CONCURRENCY` | Soft cap on concurrent event webhook deliveries per org across workers; use the same value on every worker | `128` | No |
 | `OMNARA_WORKER_CAPACITY` | Maximum agent turns executed concurrently by each worker process | `4` | No |
+| `OMNARA_WORKER_CLAIM_CONCURRENCY` | Maximum claim transactions each worker process runs at once; keep it small, since claims share one queue | `4` | No |
+| `OMNARA_WORKER_CONTINUATION_MAX_MODEL_STARTS` | Additional model calls a worker may start for an agent it already owns before releasing it back to the queue; `0` releases after every step | `2` | No |
+| `OMNARA_WORKER_CONTINUATION_MAX_DURATION` | Longest a worker keeps continuing an owned agent, checked between steps; never interrupts a running step (Go duration) | `30s` | No |
 | `OMNARA_WORKER_INBOX_CAPACITY` | Maximum integration inbox receipts processed concurrently per worker process; integer from 1 through 100 | `4` | No |
 | `OMNARA_WORKER_ASYNC_TOOL_CAPACITY` | Maximum runtime-owned asynchronous tool calls in flight per worker process | `32` | No |
 | `OMNARA_WORKER_BACKGROUND_TOOL_CAPACITY` | Background tool concurrency and in-memory queue capacity per worker process | `8` | No |
 | `OMNARA_MAINTENANCE_INTERVAL` | Maintenance loop interval (Go duration) | `1s` | No |
 
+### Agent execution
+
+Agents have typed execution state maintained by the execution store. Each transition
+updates domain records and publishes the agent's execution state and wakeup in the
+same transaction. Workers claim wakeups under fenced runtime leases and may
+continue ready work within the configured continuation budget. Rollbacks discard
+all pending state and notifications.
+
 ### Integration inbox operation
 
 Inbox concurrency is independent of agent-turn concurrency. All worker activity

docs/self-hosting/deployment.mdx

@@ -25,6 +25,13 @@ Open `http://localhost:8000` and sign up — the local email driver is `console`
 
 ## Production database upgrades
 
+The execution-state migration requires a writer-stop maintenance window. Stop API,
+worker, maintenance, and daemon-report handlers before running migrations, and
+restart them together on the new release after migration succeeds. Existing agent
+history is backfilled before the migration completes. Do not allow an older binary
+to resume writing, delete runtime leases, or clear external tool ownership during
+the upgrade. Unfinished work recovers through the normal lease and retry paths.
+
 Run the release's one-shot `migrate` image before starting its API, worker, and
 maintenance services. Connect the migrator directly to the PostgreSQL writer;
 do not use RDS Proxy or transaction-pooling PgBouncer. Long-running services

internal/agentmessage/message.go

@@ -0,0 +1,33 @@
+package agentmessage
+
+import (
+	"fmt"
+	"strings"
+)
+
+func Render(displayName, key, childPublicID, kind, text string) string {
+	label := fmt.Sprintf(
+		"Subagent %q (agent_id %s, key %q)", displayName, childPublicID, key,
+	)
+	var header string
+	switch kind {
+	case "result":
+		header = label + " finished its turn:"
+	case "refused":
+		header = label + " stopped because its model refused to continue; it produced no final answer:"
+	case "content_filtered":
+		header = label + " stopped because its model output was blocked by a content filter; " +
+			"it produced no final answer:"
+	case "failed":
+		header = label + " failed:"
+	case "question":
+		header = label + " asked a question and is paused until a human answers it. " +
+			"Messaging it with send_agent_message cancels the question."
+	default:
+		header = label + ":"
+	}
+	if strings.TrimSpace(text) == "" {
+		return header
+	}
+	return header + "\n\n" + text
+}

internal/compaction/runner.go

@@ -131,7 +131,8 @@ func (r Runner) run(
 	}
 	resolved, err := r.Resolver.Resolve(ctx, selection)
 	if err != nil {
-		if ctx.Err() != nil && (errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)) {
+		if ctx.Err() != nil &&
+			(errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)) {
 			return RunResult{}, err
 		}
 		return r.recordPreSendFailure(
@@ -265,7 +266,8 @@ func (r Runner) run(
 	providerAttempt.ProviderRequestStarted = true
 	providerAttempt.Response = response
 	if err != nil {
-		if ctx.Err() != nil && (errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)) {
+		if ctx.Err() != nil &&
+			(errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)) {
 			return RunResult{}, err
 		}
 		if _, classified := model.ClassifyError(err); !classified {
@@ -401,7 +403,7 @@ func (r Runner) run(
 		},
 	)
 	if err != nil {
-		if errors.Is(err, executionstore.ErrCheckpointBoundaryUnsafe) {
+		if errors.Is(err, storeerr.ErrCheckpointBoundaryUnsafe) {
 			nextEnd, adjustmentErr := r.nextSmallerSourceEnd(ctx, input.Plan)
 			if adjustmentErr != nil {
 				return RunResult{}, errors.Join(err, adjustmentErr)
@@ -509,7 +511,10 @@ func (r Runner) validate(input RunInput) error {
 	return nil
 }
 
-func (r Runner) resultForUnclaimed(ctx context.Context, claim executionstore.ModelCallClaim) (RunResult, error) {
+func (r Runner) resultForUnclaimed(
+	ctx context.Context,
+	claim executionstore.ModelCallClaim,
+) (RunResult, error) {
 	result := RunResult{
 		ModelCallContextID: claim.Context.ID,
 	}

internal/compaction/runner_candidate_test.go

@@ -9,6 +9,7 @@ import (
 	"github.com/omnara-ai/omnara/internal/events"
 	"github.com/omnara-ai/omnara/internal/model"
 	"github.com/omnara-ai/omnara/internal/storage/executionstore"
+	"github.com/omnara-ai/omnara/internal/storage/storeerr"
 	"github.com/stretchr/testify/require"
 )
 
@@ -114,7 +115,7 @@ func TestRunnerReplansWhenPublishFindsUnsafeBoundary(t *testing.T) {
 			textCompactionEvent(1, strings.Repeat("first closed event ", 20)),
 			textCompactionEvent(2, strings.Repeat("second closed event ", 20)),
 		},
-		publishErrs: []error{executionstore.ErrCheckpointBoundaryUnsafe},
+		publishErrs: []error{storeerr.ErrCheckpointBoundaryUnsafe},
 	}
 	client := &summaryModel{results: []summaryResult{
 		{response: completeSummaryResponse("Concise first summary.")},
@@ -307,8 +308,10 @@ func TestSafeCompactionSourceEndsRequireModelVisibleSemantics(t *testing.T) {
 }
 
 func TestRenderEventSourceReducesProjectionWithoutMutatingCanonicalOutput(t *testing.T) {
-	original := json.RawMessage(`[{"type":"structured_data","value":{"stdout":"` + strings.Repeat("x", 40_960) +
-		`TAIL"}}]`)
+	original := json.RawMessage(
+		`[{"type":"structured_data","value":{"stdout":"` + strings.Repeat("x", 40_960) +
+			`TAIL"}}]`,
+	)
 	event := mustCompactionEvent(1, "tool_result", "completed", append(json.RawMessage(nil), original...))
 	event.ToolName = "run_command"
 	event.ProviderCallID = "call_1"
@@ -446,8 +449,10 @@ func TestCompactionAllowanceFitPrefersWholeTurnAndPreservesPartialProgress(t *te
 	}{
 		{name: "whole turn fits reduced allowance", firstText: "short request", secondText: largeText,
 			firstKind: string(events.KindAgentInput), wantEnd: 2, wantReduced: true},
-		{name: "oversized turn keeps preferred partial", firstText: "short request", secondText: largeText + largeText,
-			firstKind: string(events.KindAgentInput), wantEnd: 1},
+		{name: "oversized turn keeps preferred partial",
+			firstText:  "short request",
+			secondText: largeText + largeText,
+			firstKind:  string(events.KindAgentInput), wantEnd: 1},
 		{name: "oversized turn keeps reduced partial", firstText: largeText, secondText: largeText,
 			firstKind: string(events.KindModelOutput), wantEnd: 1, wantReduced: true},
 		{name: "preferred whole turn wins", firstText: "small output", secondText: largeText,
@@ -497,7 +502,10 @@ func (m summaryModelWithExplicitCapabilities) Capabilities() model.Capabilities
 func TestCompactionRetainsOverflowLocator(t *testing.T) {
 	path := "/artifacts/art_overflow"
 	parts, err := json.Marshal([]map[string]any{
-		{"type": "structured_data", "value": map[string]any{"path": path, "preview": strings.Repeat("x", 8192)}},
+		{
+			"type":  "structured_data",
+			"value": map[string]any{"path": path, "preview": strings.Repeat("x", 8192)},
+		},
 		{"type": "text", "text": strings.Repeat("other content", 5000)},
 	})
 	if err != nil {

internal/config/config.go

@@ -66,7 +66,10 @@ type Config struct {
 	AllowInsecureDev                  bool
 	WorkerEventWebhookConcurrency     int
 	EventWebhookPerOrgConcurrency     int
+	WorkerContinuationMaxModelStarts  int
+	WorkerContinuationMaxDuration     time.Duration
 	WorkerCapacity                    int
+	WorkerClaimConcurrency            int
 	WorkerInboxCapacity               int
 	WorkerDiscordCapacity             int
 	WorkerAsyncToolCapacity           int
@@ -214,10 +217,22 @@ func Load() (Config, error) {
 	if err != nil {
 		return Config{}, err
 	}
+	workerContinuationMaxModelStarts, err := getenvInt("OMNARA_WORKER_CONTINUATION_MAX_MODEL_STARTS", 2)
+	if err != nil {
+		return Config{}, err
+	}
+	workerContinuationMaxDuration, err := time.ParseDuration(getenv("OMNARA_WORKER_CONTINUATION_MAX_DURATION", "30s"))
+	if err != nil {
+		return Config{}, fmt.Errorf("parse OMNARA_WORKER_CONTINUATION_MAX_DURATION: %w", err)
+	}
 	workerCapacity, err := getenvInt("OMNARA_WORKER_CAPACITY", 4)
 	if err != nil {
 		return Config{}, err
 	}
+	workerClaimConcurrency, err := getenvInt("OMNARA_WORKER_CLAIM_CONCURRENCY", 4)
+	if err != nil {
+		return Config{}, err
+	}
 	workerInboxCapacity, err := getenvInt("OMNARA_WORKER_INBOX_CAPACITY", 4)
 	if err != nil {
 		return Config{}, err
@@ -266,7 +281,10 @@ func Load() (Config, error) {
 		AllowInsecureDev:                  os.Getenv("OMNARA_ALLOW_INSECURE_DEV_DEFAULTS") == "1",
 		WorkerEventWebhookConcurrency:     workerEventWebhookConcurrency,
 		EventWebhookPerOrgConcurrency:     eventWebhookPerOrgConcurrency,
+		WorkerContinuationMaxModelStarts:  workerContinuationMaxModelStarts,
+		WorkerContinuationMaxDuration:     workerContinuationMaxDuration,
 		WorkerCapacity:                    workerCapacity,
+		WorkerClaimConcurrency:            workerClaimConcurrency,
 		WorkerInboxCapacity:               workerInboxCapacity,
 		WorkerDiscordCapacity:             workerDiscordCapacity,
 		WorkerAsyncToolCapacity:           workerAsyncToolCapacity,
@@ -622,9 +640,18 @@ func (cfg Config) ValidateWorker() error {
 	if cfg.EventWebhookPerOrgConcurrency <= 0 {
 		return fmt.Errorf("OMNARA_EVENT_WEBHOOK_PER_ORG_CONCURRENCY must be positive")
 	}
+	if cfg.WorkerContinuationMaxModelStarts < 0 {
+		return fmt.Errorf("OMNARA_WORKER_CONTINUATION_MAX_MODEL_STARTS must be non-negative")
+	}
+	if cfg.WorkerContinuationMaxDuration <= 0 {
+		return fmt.Errorf("OMNARA_WORKER_CONTINUATION_MAX_DURATION must be positive")
+	}
 	if cfg.WorkerCapacity <= 0 {
 		return fmt.Errorf("OMNARA_WORKER_CAPACITY must be positive")
 	}
+	if cfg.WorkerClaimConcurrency <= 0 {
+		return fmt.Errorf("OMNARA_WORKER_CLAIM_CONCURRENCY must be positive")
+	}
 	if cfg.WorkerInboxCapacity < 1 || cfg.WorkerInboxCapacity > 100 {
 		return fmt.Errorf("OMNARA_WORKER_INBOX_CAPACITY must be between 1 and 100")
 	}