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

.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
@@ -32,6 +33,7 @@ INTEGRATION_STORAGE_PACKAGES := \
 	./internal/storage/executionstore \
 	./internal/storage/identitystore \
 	./internal/storage/integrationstore \
+	./internal/storage/internal/agentexecution \
 	./internal/testutil/storagetest
 INTEGRATION_HTTPAPI_PACKAGES := \
 	./internal/httpapi \
@@ -259,6 +261,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 +270,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/events/webhooks.mdx

@@ -6,6 +6,8 @@ icon: "webhook"
 
 Configure an [event webhook](/agents/configuration#event-webhook) to receive JSON POST requests with `event` and `data`. The event name determines the data schema. These are the same data objects used by the [event stream](/events/streaming).
 
+Webhook deliveries are suppressed for deleted projects and organizations, including events produced by their deletion.
+
 Every request includes `Webhook-Id` (stable across retries) and `Webhook-Timestamp` (Unix seconds for that attempt). Signed requests also include `Webhook-Signature`. Verify the signature against the raw request body before parsing it, using [Standard Webhooks](https://github.com/standard-webhooks/standard-webhooks/blob/main/spec/standard-webhooks.md).
 
 ## Receive and verify

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")
 	}

internal/config/config_test.go

@@ -1106,6 +1106,35 @@ func TestValidateWorkerLoadsExplicitWorkerCapacity(t *testing.T) {
 	}
 }
 
+func TestValidateWorkerLoadsExplicitClaimConcurrency(t *testing.T) {
+	t.Setenv("OMNARA_ALLOW_INSECURE_DEV_DEFAULTS", "1")
+	t.Setenv("OMNARA_WORKER_CLAIM_CONCURRENCY", "2")
+
+	cfg, err := Load()
+	if err != nil {
+		t.Fatalf("load config: %v", err)
+	}
+	if err := cfg.ValidateWorker(); err != nil {
+		t.Fatalf("validate worker: %v", err)
+	}
+	if cfg.WorkerClaimConcurrency != 2 {
+		t.Fatalf("unexpected claim concurrency: %+v", cfg)
+	}
+}
+
+func TestValidateWorkerRejectsInvalidClaimConcurrency(t *testing.T) {
+	t.Setenv("OMNARA_ALLOW_INSECURE_DEV_DEFAULTS", "1")
+	t.Setenv("OMNARA_WORKER_CLAIM_CONCURRENCY", "0")
+
+	cfg, err := Load()
+	if err != nil {
+		t.Fatalf("load config: %v", err)
+	}
+	if err := cfg.ValidateWorker(); err == nil {
+		t.Fatal("expected invalid claim concurrency error")
+	}
+}
+
 func TestValidateWorkerLoadsExplicitAsyncToolCapacity(t *testing.T) {
 	t.Setenv("OMNARA_ALLOW_INSECURE_DEV_DEFAULTS", "1")
 	t.Setenv("OMNARA_WORKER_ASYNC_TOOL_CAPACITY", "64")