diff --git a/.env.example b/.env.example index a08225874..a20a6bae6 100644 --- a/.env.example +++ b/.env.example @@ -10,6 +10,7 @@ OMNARA_ALLOW_INSECURE_DEV_DEFAULTS=1 OMNARA_WORKER_EVENT_WEBHOOK_CONCURRENCY=128 OMNARA_EVENT_WEBHOOK_PER_ORG_CONCURRENCY=128 OMNARA_WORKER_CAPACITY=4 +OMNARA_WORKER_CLAIM_CONCURRENCY=16 OMNARA_WORKER_INBOX_CAPACITY=4 OMNARA_WORKER_DISCORD_CAPACITY=1024 OMNARA_WORKER_ASYNC_TOOL_CAPACITY=32 diff --git a/.golangci.yml b/.golangci.yml index 372df6376..054337145 100644 --- a/.golangci.yml +++ b/.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 diff --git a/Makefile b/Makefile index 8d702e85f..97418e3bb 100644 --- a/Makefile +++ b/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 diff --git a/api/openapi/openapi.yaml b/api/openapi/openapi.yaml index 1f1d011a7..2ec4322e7 100644 --- a/api/openapi/openapi.yaml +++ b/api/openapi/openapi.yaml @@ -9138,6 +9138,7 @@ components: - internal_error - service_unavailable - idempotency_key_conflict + - agent_backlog_full - state_transition_conflict - managed_work_admission_denied - pending_work @@ -9168,6 +9169,7 @@ components: - internal_error - service_unavailable - idempotency_key_conflict + - agent_backlog_full - state_transition_conflict - managed_work_admission_denied - pending_work @@ -9192,6 +9194,7 @@ components: - expected_digest_required - not_a_file - idempotency_key_conflict + - agent_backlog_full - state_transition_conflict - pending_work - not_wake_capable diff --git a/cmd/maintenance/execution_head.go b/cmd/maintenance/execution_head.go new file mode 100644 index 000000000..1f0fecdf2 --- /dev/null +++ b/cmd/maintenance/execution_head.go @@ -0,0 +1,74 @@ +package main + +import ( + "context" + "log/slog" + "time" + + "github.com/google/uuid" + "github.com/omnara-ai/omnara/internal/storage/executionstore" + "github.com/omnara-ai/omnara/observability/metrics" +) + +const ( + executionHeadInterval = 5 * time.Minute + executionHeadSampleSize = 10 + executionHeadJob = "execution_head_drift" +) + +func runExecutionHeadLoop(ctx context.Context, log *slog.Logger, recorder *metrics.JobRecorder, + sample func(context.Context, uuid.UUID, int32) (executionstore.ExecutionHeadSampleReport, error)) { + recorder.InitializeItems(executionHeadJob, "checked", "mismatch", "error") + var cursor uuid.UUID + for { + cursor = runExecutionHeadTick(ctx, log, recorder, cursor, sample) + timer := time.NewTimer(jitteredMaintenanceDelay(executionHeadInterval)) + select { + case <-ctx.Done(): + if !timer.Stop() { + <-timer.C + } + return + case <-timer.C: + } + } +} + +func runExecutionHeadTick(ctx context.Context, log *slog.Logger, recorder *metrics.JobRecorder, cursor uuid.UUID, + sample func(context.Context, uuid.UUID, int32) (executionstore.ExecutionHeadSampleReport, error)) uuid.UUID { + started := time.Now() + report, err := sample(ctx, cursor, executionHeadSampleSize) + result := metrics.JobResultSuccess + var checked, mismatch, failed int64 + for _, sample := range report.Samples { + if sample.Err != nil { + failed++ + log.Error("sample execution head", "agent_id", sample.AgentID, "project_id", sample.ProjectID, "error", sample.Err) + continue + } + checked++ + if len(sample.Differences) > 0 { + mismatch++ + fields := make([]string, 0, len(sample.Differences)) + for _, difference := range sample.Differences { + fields = append(fields, difference.Field) + } + log.Error("execution head drift", "agent_id", sample.AgentID, "project_id", sample.ProjectID, + "fields", fields, "differences", sample.Differences) + } + } + if err != nil || failed > 0 { + result = metrics.JobResultError + } + if ctx.Err() != nil { + result = metrics.JobResultCanceled + } + if err != nil && ctx.Err() == nil { + log.Error("sample execution heads", "error", err) + } + recorder.RecordItems(executionHeadJob, "checked", checked) + recorder.RecordItems(executionHeadJob, "mismatch", mismatch) + recorder.RecordItems(executionHeadJob, "error", failed) + recorder.RecordRun(executionHeadJob, result, time.Since(started)) + return report.Cursor +} diff --git a/cmd/maintenance/execution_head_test.go b/cmd/maintenance/execution_head_test.go new file mode 100644 index 000000000..4879d1db6 --- /dev/null +++ b/cmd/maintenance/execution_head_test.go @@ -0,0 +1,109 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "log/slog" + "net/http" + "net/http/httptest" + "testing" + + "github.com/google/uuid" + "github.com/omnara-ai/omnara/internal/storage/executionstore" + "github.com/omnara-ai/omnara/observability/metrics" + "github.com/stretchr/testify/require" +) + +func TestExecutionHeadTickOutcomes(t *testing.T) { + for _, outcome := range []string{"success", "error", "canceled"} { + t.Run(outcome, func(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + var logs bytes.Buffer + set := metrics.New() + recorder := metrics.NewJobRecorder(set, metrics.SubsystemMaintenance) + runExecutionHeadTick(ctx, slog.New(slog.NewJSONHandler(&logs, nil)), recorder, uuid.Nil, + func(context.Context, uuid.UUID, int32) (executionstore.ExecutionHeadSampleReport, error) { + report := executionstore.ExecutionHeadSampleReport{ + Samples: []executionstore.ExecutionHeadSample{{AgentID: uuid.New(), ProjectID: uuid.New()}}} + switch outcome { + case "error": + report.Samples[0].Err = errors.New("timeout") + case "canceled": + cancel() + return report, context.Canceled + } + return report, nil + }) + + response := httptest.NewRecorder() + set.Handler().ServeHTTP(response, httptest.NewRequest(http.MethodGet, "/metrics", nil)) + require.Contains(t, response.Body.String(), + `omnara_maintenance_job_runs_total{job="execution_head_drift",result="`+outcome+`"} 1`) + }) + } +} + +func TestExecutionHeadTickMismatch(t *testing.T) { + var sample executionstore.ExecutionHeadSample + require.NoError(t, json.Unmarshal( + []byte(`{"Differences":[{"field":"stop_sequence","stored":8,"rebuilt":3}]}`), &sample)) + sample.AgentID = uuid.New() + sample.ProjectID = uuid.New() + set := metrics.New() + var logs bytes.Buffer + runExecutionHeadTick(t.Context(), slog.New(slog.NewJSONHandler(&logs, nil)), + metrics.NewJobRecorder(set, metrics.SubsystemMaintenance), uuid.Nil, + func(context.Context, uuid.UUID, int32) (executionstore.ExecutionHeadSampleReport, error) { + return executionstore.ExecutionHeadSampleReport{ + Samples: []executionstore.ExecutionHeadSample{sample}}, nil + }) + require.Contains(t, logs.String(), `"level":"ERROR"`) + require.Contains(t, logs.String(), sample.AgentID.String()) + require.Contains(t, logs.String(), sample.ProjectID.String()) + require.Contains(t, logs.String(), `"fields":["stop_sequence"]`) + require.Contains(t, logs.String(), `"stored":8,"rebuilt":3`) + response := httptest.NewRecorder() + set.Handler().ServeHTTP(response, httptest.NewRequest(http.MethodGet, "/metrics", nil)) + for _, outcome := range []string{"checked", "mismatch"} { + require.Contains(t, response.Body.String(), + `omnara_maintenance_job_items_total{job="execution_head_drift",outcome="`+outcome+`"} 1`) + } +} + +func TestExecutionHeadTickCursor(t *testing.T) { + var logs bytes.Buffer + var cursor uuid.UUID + for range 3 { + previous, next := cursor, uuid.New() + cursor = runExecutionHeadTick(t.Context(), slog.New(slog.NewJSONHandler(&logs, nil)), nil, cursor, + func(_ context.Context, after uuid.UUID, n int32) (executionstore.ExecutionHeadSampleReport, error) { + require.Equal(t, previous, after) + require.EqualValues(t, 10, n) + return executionstore.ExecutionHeadSampleReport{Cursor: next}, nil + }) + require.Equal(t, next, cursor) + } +} + +func TestExecutionHeadLoopInitializesItems(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + set := metrics.New() + var logs bytes.Buffer + runExecutionHeadLoop(ctx, slog.New(slog.NewJSONHandler(&logs, nil)), + metrics.NewJobRecorder(set, metrics.SubsystemMaintenance), + func(_ context.Context, after uuid.UUID, _ int32) (executionstore.ExecutionHeadSampleReport, error) { + require.Equal(t, uuid.Nil, after) + response := httptest.NewRecorder() + set.Handler().ServeHTTP(response, httptest.NewRequest(http.MethodGet, "/metrics", nil)) + for _, outcome := range []string{"checked", "mismatch", "error"} { + require.Contains(t, response.Body.String(), + `omnara_maintenance_job_items_total{job="execution_head_drift",outcome="`+outcome+`"} 0`) + } + cancel() + return executionstore.ExecutionHeadSampleReport{}, context.Canceled + }) +} diff --git a/cmd/maintenance/main.go b/cmd/maintenance/main.go index 52d0c5ff0..699459005 100644 --- a/cmd/maintenance/main.go +++ b/cmd/maintenance/main.go @@ -166,6 +166,12 @@ func main() { defer close(memoryCleanupDone) runMemoryCleanupLoop(ctx, logger, memoryCleanupInterval, store.Memories().CleanupFiles) }() + executionHeadDone := make(chan struct{}) + jobRecorder := metrics.NewJobRecorder(metricSet, metrics.SubsystemMaintenance) + go func() { + defer close(executionHeadDone) + runExecutionHeadLoop(ctx, logger, jobRecorder, store.Execution().SampleExecutionHeads) + }() runtimeDiscoveryDone := make(chan struct{}) go func() { defer close(runtimeDiscoveryDone) @@ -241,6 +247,7 @@ func main() { <-machineLoopDone <-idleAgentArchiveLoopDone <-memoryCleanupDone + <-executionHeadDone <-runtimeDiscoveryDone <-runtimeRecheckDone <-discordRuntimeDone diff --git a/cmd/worker/main.go b/cmd/worker/main.go index 62fa8d46d..135ab21a5 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -241,8 +241,10 @@ func main() { StreamLog: log, }) kernelWorker := workerpkg.NewWorker(store.Execution(), executor, workerpkg.Options{ + Metrics: metrics.NewWorkerRecorder(metricSet), Log: log, Capacity: cfg.WorkerCapacity, + ClaimConcurrency: cfg.WorkerClaimConcurrency, AsyncToolCapacity: cfg.WorkerAsyncToolCapacity, ControlSubscriber: redisBus, }) diff --git a/compose.yaml b/compose.yaml index 39f5fd094..1f8860993 100644 --- a/compose.yaml +++ b/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:-16} 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} diff --git a/docs/api-reference/openapi.yaml b/docs/api-reference/openapi.yaml index 026d728c5..df2155854 100644 --- a/docs/api-reference/openapi.yaml +++ b/docs/api-reference/openapi.yaml @@ -9139,6 +9139,7 @@ components: - internal_error - service_unavailable - idempotency_key_conflict + - agent_backlog_full - state_transition_conflict - managed_work_admission_denied - pending_work @@ -9169,6 +9170,7 @@ components: - internal_error - service_unavailable - idempotency_key_conflict + - agent_backlog_full - state_transition_conflict - managed_work_admission_denied - pending_work @@ -9193,6 +9195,7 @@ components: - expected_digest_required - not_a_file - idempotency_key_conflict + - agent_backlog_full - state_transition_conflict - pending_work - not_wake_capable diff --git a/docs/api/overview.mdx b/docs/api/overview.mdx index 26411dd23..9e65fa78a 100644 --- a/docs/api/overview.mdx +++ b/docs/api/overview.mdx @@ -67,7 +67,9 @@ Pass the returned cursor back on the next request; a `null` cursor means you've ## Idempotency -Create endpoints — orgs, projects, profiles, inputs, machines, and machine/pool grants — accept an optional `Idempotency-Key` header (the API reference marks each one). Retrying with the same key and an identical body returns the original resource instead of creating a duplicate — a replay returns `200` where the first request returned `201`; a conflicting retry returns `409` with code `idempotency_key_conflict`. +Create endpoints — orgs, projects, profiles, machines, and machine/pool grants — accept an optional `Idempotency-Key` header (the API reference marks each one). Retrying with the same key and an identical body returns the original resource instead of creating a duplicate — a replay returns `200` where the first request returned `201`; a conflicting retry returns `409` with code `idempotency_key_conflict`. + +Agent input creation and config changes use the key alone: a retry returns the original stored result without comparing the request body. Organization creation also accepts this header but does not require it. A key determines the proposed organization ID, so a failed retry uses the same ID. Once the organization commits, reusing the key returns it without another hosted call. @@ -84,7 +86,7 @@ Branch on `code` — it's a stable contract, enumerated in the OpenAPI spec, and Two rules worth internalizing: - **Resources you cannot read return `404`, not `403`.** The API doesn't confirm the existence of things outside your access. A `403` always means "the resource is visible to you, but your role doesn't permit this action." -- **Retry `424`, `429` and `5xx` with backoff, reusing the same `Idempotency-Key`** so a retry can never act twice. `424` with code `upstream_unavailable` means a server you configured, such as an MCP server, did not respond or failed; retries succeed once that server recovers. A `409` from MCP tool discovery means another refresh of the same server or auth secret is in progress; retry it shortly. Other `4xx` errors won't change on retry — fix the request instead. +- **Retry `424`, `429` and `5xx` with backoff, reusing the same `Idempotency-Key`** so a retry can never act twice. `424` with code `upstream_unavailable` means a server you configured, such as an MCP server, did not respond or failed; retries succeed once that server recovers. A `409` from MCP tool discovery means another refresh of the same server or auth secret is in progress; retry it shortly. A `409` with code `agent_backlog_full` means the agent has 50 pending messages; wait for processing or cancel a pending message before retrying. Other `4xx` errors won't change on retry — fix the request instead. ## Request size limits diff --git a/docs/events/sending-input.mdx b/docs/events/sending-input.mdx index 08195f69f..f9fa3ba78 100644 --- a/docs/events/sending-input.mdx +++ b/docs/events/sending-input.mdx @@ -29,13 +29,15 @@ export AGENT="agt_v4qtrwz3jehcyd5n6a2bfgik7m" ## Send a message +An agent accepts up to 50 pending messages across queued and steering inputs; further submissions return `409` with code `agent_backlog_full` until messages are processed or canceled, while child lifecycle reports are always delivered. + The examples assume the client, `orgID`, and `projectID` setup from the [quickstart](/quickstart), with the `$AGENT` ID above available as `agentID` for the SDK. The CLI has no command for sending input — use the REST API or the SDK. - `state: received` means Omnara accepted the input. When the agent processes it, the state becomes `resolved` and a matching event appears on the timeline. An input can instead become `canceled` or `rejected` if it is withdrawn or cannot be delivered. + `state: received` means Omnara accepted the input. When the agent processes it, the state becomes `resolved` and a matching event appears on the timeline. An input can instead become `canceled` if it is withdrawn before the agent processes it. If your system retries requests, send an `Idempotency-Key` to avoid delivering the same message twice. diff --git a/docs/events/webhooks.mdx b/docs/events/webhooks.mdx index 8e3ec548c..efdfde679 100644 --- a/docs/events/webhooks.mdx +++ b/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 diff --git a/docs/self-hosting/configuration.mdx b/docs/self-hosting/configuration.mdx index d206b9cf1..abab2dfa3 100644 --- a/docs/self-hosting/configuration.mdx +++ b/docs/self-hosting/configuration.mdx @@ -28,11 +28,19 @@ 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` | Concurrent claim loops per worker process | `16` | 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 | +### Monitoring agent execution + +Alert on these counters: + +- `increase(omnara_maintenance_job_items_total{job="execution_head_drift",outcome="mismatch"}[1h]) > 0`: an agent's stored execution state disagrees with its history. +- `increase(omnara_worker_claim_deferrals_total[15m]) > 0`: a worker could not claim an agent and retried it a minute later. Worker error logs name the agent. + ### Integration inbox operation Inbox concurrency is independent of agent-turn concurrency. All worker activity diff --git a/frontend/apps/web/src/components/agents/AgentComposer.test.tsx b/frontend/apps/web/src/components/agents/AgentComposer.test.tsx new file mode 100644 index 000000000..72ac08e73 --- /dev/null +++ b/frontend/apps/web/src/components/agents/AgentComposer.test.tsx @@ -0,0 +1,161 @@ +/** @vitest-environment happy-dom */ + +import type { UseAgentChatResult } from '@omnara/react' +import { ApiError } from '@omnara/sdk' +import { act, useState } from 'react' +import { createRoot } from 'react-dom/client' +import { expect, it, vi } from 'vitest' + +import { MessageScrollerProvider } from '@/components/ui/message-scroller' +import { enableReactActEnvironment } from '@/test/react-act' + +import { AgentComposer } from './AgentComposer' + +const message = + 'This agent already has 50 pending messages. Wait for messages to be processed or cancel a pending message, then try again.' + +type Send = UseAgentChatResult['sendMessage'] + +interface InFlightSend { + settle?: (failure?: Error) => void +} + +function Composer({ send }: { send: Send }) { + const [error, setError] = useState() + const chat: UseAgentChatResult = { + events: [], + messages: [], + status: 'ready', + isWorking: false, + error, + streamError: undefined, + historyStatus: 'success', + historyError: null, + retryHistory: vi.fn(), + reconnect: vi.fn(), + hasOlderMessages: false, + isLoadingOlderMessages: false, + loadOlderMessages: vi.fn(), + inputBacklog: { + inputs: [], + actionPending: false, + beginCancellation: vi.fn(), + cancel: vi.fn(), + promote: vi.fn(), + move: vi.fn(), + }, + sendMessage: async (input) => { + setError(undefined) + try { + await send(input) + } catch (cause) { + setError(cause instanceof Error ? cause : new Error(String(cause))) + throw cause + } + }, + } + return ( +