From 0264b54f5fe3b5d2fa68f6e773c68eaf70222552 Mon Sep 17 00:00:00 2001 From: SaladDay <1203511142@qq.com> Date: Mon, 28 Sep 2026 09:35:04 +0000 Subject: [PATCH 1/5] Observe deployment model defaults after committed root Turns --- CONTRIBUTING.md | 20 + .../core/deployment-model-providers.mdx | 9 +- apps/docs/openapi/core-api.yaml | 31 +- apps/docs/openapi/sources.json | 6 +- contracts/agents-api/core.openapi.yaml | 28 +- contracts/agents-api/model-execution.md | 27 ++ .../agents-client/src/admin-client.test.ts | 2 +- .../agents-client/src/admin-projection.ts | 13 +- .../src/provider-observations.test.ts | 16 + packages/agents-client/src/types.ts | 5 + services/agents-api/internal/api/handler.go | 12 +- .../internal/api/harness_model_providers.go | 25 +- .../api/model_provider_fixture_test.go | 6 +- .../internal/api/session_model_defaults.go | 28 +- .../db/queries/deployment_model_providers.sql | 13 +- .../deployment_provider_observations.sql | 34 ++ .../session_execution_configuration.sql | 4 +- .../db/sqlc/deployment_model_providers.sql.go | 52 ++- .../deployment_provider_observations.sql.go | 62 +++ .../agents-api/internal/db/sqlc/models.go | 10 +- .../session_execution_configuration.sql.go | 11 +- .../deployment_provider_observations.go | 34 ++ .../deployment_provider_observations_test.go | 219 +++++++++ .../internal/execution/dispatcher.go | 5 +- .../agent_execution_defaults_http_test.go | 4 +- .../store/deployment_model_providers.go | 40 +- .../deployment_model_providers_http_test.go | 76 ++++ .../store/deployment_model_providers_test.go | 4 +- ...ployment_provider_observation_lock_test.go | 118 +++++ .../store/deployment_provider_observations.go | 49 ++ .../deployment_provider_observations_test.go | 424 ++++++++++++++++++ .../store/model_provider_fixture_test.go | 6 +- .../store/session_execution_configuration.go | 9 +- .../internal/store/session_initial_input.go | 4 +- .../agents-api/internal/store/sessions.go | 28 +- ...00084_deployment_provider_observations.sql | 27 ++ 36 files changed, 1350 insertions(+), 111 deletions(-) create mode 100644 packages/agents-client/src/provider-observations.test.ts create mode 100644 services/agents-api/internal/db/queries/deployment_provider_observations.sql create mode 100644 services/agents-api/internal/db/sqlc/deployment_provider_observations.sql.go create mode 100644 services/agents-api/internal/execution/deployment_provider_observations.go create mode 100644 services/agents-api/internal/execution/deployment_provider_observations_test.go create mode 100644 services/agents-api/internal/store/deployment_provider_observation_lock_test.go create mode 100644 services/agents-api/internal/store/deployment_provider_observations.go create mode 100644 services/agents-api/internal/store/deployment_provider_observations_test.go create mode 100644 services/agents-api/migrations/000084_deployment_provider_observations.sql diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index bec66792d..ab0832e3c 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -2502,6 +2502,26 @@ replaced; do not carry obsolete compatibility code forward to satisfy this secti retries return committed state without replay. No Turn-level overrides or provider catalog is included. Public input/null semantics and examples live in `contracts/agents-api/model-execution.md`. +- Deployment-default observations use a private UUID generated on every PUT, + including identical replacements. Read its ciphertext/revision together and freeze + the pair only at new Session creation; retries and historical Sessions never gain + or replace revision metadata. After a successful root terminal commit, one + independent pool operation has at most one second to update the matching current + revision. SQL verifies tenant, root Turn and committed outcome; only completed + Turns and fixed native provider failures with authoritative `engine_failed` qualify. + Never observe cancelled work, Core/runtime errors or input-policy classifications. + The metadata-only transaction sets server statement/lock timeouts within the + remaining client budget and issues one UPDATE, so a client timeout cannot leave + an indefinitely waiting statement. The update locks only the default and samples + DB receipt time after the lock. + Errors throttle for 30 seconds regardless of code; ordinary successes throttle + for 30 seconds, with one immediate recovery write after each accepted error. + An unchanged revision has at most three effective writes in any half-open + 30-second interval of nondecreasing DB time. Clock rollback may suppress ordinary + writes; private recovery state permits only one recovery without clamping time. + Observations never change `updated_at`, readiness or execution truth. They can be + lost or stay stale indefinitely without another eligible Turn; there is no queue, + retry, probe, history backfill or provider text in the safe fields. - Session execution-configuration reads use a separate immutable safe projection, written with provenance in the same creation transaction as Session resources. Read no credential ciphertext and never recompute sources from current Agents diff --git a/apps/docs/content/docs/api-reference/core/deployment-model-providers.mdx b/apps/docs/content/docs/api-reference/core/deployment-model-providers.mdx index 5f8f685ff..4859043f6 100644 --- a/apps/docs/content/docs/api-reference/core/deployment-model-providers.mdx +++ b/apps/docs/content/docs/api-reference/core/deployment-model-providers.mdx @@ -18,7 +18,11 @@ _openapi: never returned; api_key_configured reports that one is set. - content: >- Core key only. Returns the safe view; the key is never returned. 404 - when the harness does not exist or has no deployment default. + when the harness does not exist or has no deployment default. Nullable + last_used_at, last_error_code and last_error_at are best-effort + observations of committed root Turns using this exact default + revision; they do not establish current readiness and may remain stale + indefinitely. - content: >- Core key only. Idempotent; each successful request is audited. Sessions that already froze the default keep it. Afterwards new @@ -35,7 +39,8 @@ _openapi: bundle freeze this default into their encrypted snapshot; existing Sessions never change. self_hosted Sessions never use it. The key is encrypted and never returned. Each write records an administrator - audit entry without the key. + audit entry without the key and resets last_used_at, last_error_code + and last_error_at to null, including identical writes. --- {/* This file was generated by Fumadocs. Do not edit this file directly. Any changes should be made by running the generation command again. */} diff --git a/apps/docs/openapi/core-api.yaml b/apps/docs/openapi/core-api.yaml index 60db44d0c..e27a325ae 100644 --- a/apps/docs/openapi/core-api.yaml +++ b/apps/docs/openapi/core-api.yaml @@ -152,7 +152,7 @@ paths: tags: - Deployment Model Providers get: - description: Core key only. Returns the safe view; the key is never returned. 404 when the harness does not exist or has no deployment default. + description: Core key only. Returns the safe view; the key is never returned. 404 when the harness does not exist or has no deployment default. Nullable last_used_at, last_error_code and last_error_at are best-effort observations of committed root Turns using this exact default revision; they do not establish current readiness and may remain stale indefinitely. parameters: - schema: type: string @@ -195,7 +195,7 @@ paths: tags: - Deployment Model Providers put: - description: 'Core key only. The body is the complete x_agents_core.model_provider bundle, including the write-only api_key; there is no partial update and bundles are never merged. The provider is validated for this harness: an HTTPS base_url without credentials, query or fragment, the harness''s protocol (responses for codex, anthropic for claude_sdk and mcode) and, for mcode, positive context_window and max_output_tokens. New openai_hosted and none Sessions that resolve no Session or Agent bundle freeze this default into their encrypted snapshot; existing Sessions never change. self_hosted Sessions never use it. The key is encrypted and never returned. Each write records an administrator audit entry without the key.' + description: 'Core key only. The body is the complete x_agents_core.model_provider bundle, including the write-only api_key; there is no partial update and bundles are never merged. The provider is validated for this harness: an HTTPS base_url without credentials, query or fragment, the harness''s protocol (responses for codex, anthropic for claude_sdk and mcode) and, for mcode, positive context_window and max_output_tokens. New openai_hosted and none Sessions that resolve no Session or Agent bundle freeze this default into their encrypted snapshot; existing Sessions never change. self_hosted Sessions never use it. The key is encrypted and never returned. Each write records an administrator audit entry without the key and resets last_used_at, last_error_code and last_error_at to null, including identical writes.' parameters: - schema: type: string @@ -4454,6 +4454,30 @@ components: - codex - mcode type: string + last_error_at: + anyOf: + - format: date-time + type: string + - type: 'null' + last_error_code: + anyOf: + - enum: + - authentication_error + - connection_failed + - rate_limit_exceeded + - usage_limit_exceeded + - server_overloaded + - server_error + - resource_not_found + - request_timeout + - invalid_request + type: string + - type: 'null' + last_used_at: + anyOf: + - format: date-time + type: string + - type: 'null' max_output_tokens: type: integer object: @@ -4471,6 +4495,9 @@ components: - api_key_configured - base_url - harness + - last_error_at + - last_error_code + - last_used_at - object - protocol - updated_at diff --git a/apps/docs/openapi/sources.json b/apps/docs/openapi/sources.json index 49e202ccd..bf2a7c62a 100644 --- a/apps/docs/openapi/sources.json +++ b/apps/docs/openapi/sources.json @@ -1,7 +1,7 @@ { "sources": { "contracts/agents-api/openapi.yaml": "a8896e8b67d78ed0608c7cc5636077ec33f4b4895a04bd6cfb8b038afe941e01", - "contracts/agents-api/core.openapi.yaml": "8fe158844c170eea78d4b24e30e7caa2adbedd58ffb4a863bdd9c98ecbb50618", + "contracts/agents-api/core.openapi.yaml": "7801fc77259893815fe9cdcb404f045b375711b9b075eab2b2ed8c5dba55e24b", "contracts/agents-api/runtime.openapi.yaml": "23852205d3b45892f2de9ecf3d52518c0dd0808962c0c512f219b565aa7ccef2" }, "outputs": { @@ -22,7 +22,7 @@ "content/docs/api-reference/meta.json": "56bf1dc145ccc8adde38c5f956f1a06f18ef18b646122250d01a294c516f8f83", "content/docs/api-reference/index.mdx": "9fd6b1c54149874d2999235067c101326f9e8a2e707a70b6672c81eb3ebfe63f", "content/docs/api-reference/core/core-administration.mdx": "5fc3ada51d2190a141f13738151e4d513390cc3bd66ce6f5013994bda68569b0", - "content/docs/api-reference/core/deployment-model-providers.mdx": "b2795fadcc05a1e7195b6d34c3f234cb29cc5e3f580d1521e31965453e654347", + "content/docs/api-reference/core/deployment-model-providers.mdx": "d2e1ed267def55e1b64fdf28fbec81f83b9c2556b1db4392ede6a19b1465b50c", "content/docs/api-reference/core/administrator-projects.mdx": "8fd687bb2a7119fa16e938d7c71710c2816afb9cf959eb6849ef5e6300c5419b", "content/docs/api-reference/core/agents.mdx": "b6c089fbea4b07bc4249e7c7ca5699d47f1512a4fc2d8ee01cb26910ae541a84", "content/docs/api-reference/core/environment-templates.mdx": "05907f35d7f02d66af9da8b1efaa3765badb4cdf7b94d0f6c3ab0413a2cfa1a9", @@ -40,7 +40,7 @@ "content/docs/api-reference/core/vaults.mdx": "12d0aac96cd4a3472ef4eaa287f7945f78862017956e79bfc3ed0ef8e772e78a", "content/docs/api-reference/core/credentials.mdx": "27d586b9a8fd738b9947802f32c80458195370252886bc26abedccf2125ee981", "content/docs/api-reference/core/sandbox-manager.mdx": "6be7c52e057365949f0247c18c710561d680847a955f78f1bbe57825d4d5f549", - "openapi/core-api.yaml": "4d72411e4fc9945aba575a91b9c95e7231262e2749f87796646ef17768022238", + "openapi/core-api.yaml": "7079694e447c0783f85aeb3b73f23af68ebb066540c67251bbae34e7e4bd8e62", "content/docs/api-reference/core/meta.json": "588b163ed2de2100b0f8376688274f80698a6bf6ac037eef56d17c93fa8a5062", "content/docs/api-reference/core/index.mdx": "ddf0a8d87a6c21761beac703a32f2a40486a1c74abb341bd726f40c44a866405", "content/docs/api-reference/machine/sandbox-node.mdx": "5b367a08d43cea82686345d2ba1c75cc90627a7b2970fe42a0df9d77fb3e1e80", diff --git a/contracts/agents-api/core.openapi.yaml b/contracts/agents-api/core.openapi.yaml index e6607611a..d8dcda463 100644 --- a/contracts/agents-api/core.openapi.yaml +++ b/contracts/agents-api/core.openapi.yaml @@ -346,6 +346,27 @@ definitions: - codex - mcode type: string + last_error_at: + format: date-time + type: string + x-nullable: true + last_error_code: + enum: + - authentication_error + - connection_failed + - rate_limit_exceeded + - usage_limit_exceeded + - server_overloaded + - server_error + - resource_not_found + - request_timeout + - invalid_request + type: string + x-nullable: true + last_used_at: + format: date-time + type: string + x-nullable: true max_output_tokens: type: integer object: @@ -363,6 +384,9 @@ definitions: - api_key_configured - base_url - harness + - last_error_at + - last_error_code + - last_used_at - object - protocol - updated_at @@ -3429,7 +3453,7 @@ paths: tags: - Deployment Model Providers get: - description: Core key only. Returns the safe view; the key is never returned. 404 when the harness does not exist or has no deployment default. + description: Core key only. Returns the safe view; the key is never returned. 404 when the harness does not exist or has no deployment default. Nullable last_used_at, last_error_code and last_error_at are best-effort observations of committed root Turns using this exact default revision; they do not establish current readiness and may remain stale indefinitely. parameters: - description: Harness enum: @@ -3467,7 +3491,7 @@ paths: put: consumes: - application/json - description: 'Core key only. The body is the complete x_agents_core.model_provider bundle, including the write-only api_key; there is no partial update and bundles are never merged. The provider is validated for this harness: an HTTPS base_url without credentials, query or fragment, the harness''s protocol (responses for codex, anthropic for claude_sdk and mcode) and, for mcode, positive context_window and max_output_tokens. New openai_hosted and none Sessions that resolve no Session or Agent bundle freeze this default into their encrypted snapshot; existing Sessions never change. self_hosted Sessions never use it. The key is encrypted and never returned. Each write records an administrator audit entry without the key.' + description: 'Core key only. The body is the complete x_agents_core.model_provider bundle, including the write-only api_key; there is no partial update and bundles are never merged. The provider is validated for this harness: an HTTPS base_url without credentials, query or fragment, the harness''s protocol (responses for codex, anthropic for claude_sdk and mcode) and, for mcode, positive context_window and max_output_tokens. New openai_hosted and none Sessions that resolve no Session or Agent bundle freeze this default into their encrypted snapshot; existing Sessions never change. self_hosted Sessions never use it. The key is encrypted and never returned. Each write records an administrator audit entry without the key and resets last_used_at, last_error_code and last_error_at to null, including identical writes.' parameters: - description: Harness enum: diff --git a/contracts/agents-api/model-execution.md b/contracts/agents-api/model-execution.md index 90c5d325d..87d4ab7b1 100644 --- a/contracts/agents-api/model-execution.md +++ b/contracts/agents-api/model-execution.md @@ -182,3 +182,30 @@ Parsar manages its own workspace catalog and encrypted keys, sends this extensio only on the first Core Session request, and retains a private encrypted snapshot for uncertain creation retries. Catalog updates and deletion affect new Sessions; existing Sessions retain their original model, endpoint and key. + +### Deployment default observations + +The Core-only harness/default-provider reads include nullable `last_used_at`, +`last_error_code` and `last_error_at`. PUT resets all three, even for the same bundle. +Completed root Turns contribute use observations only when their Session froze +that exact current default revision. Failed root Turns contribute only the fixed +native provider codes `authentication_error`, `connection_failed`, +`rate_limit_exceeded`, `usage_limit_exceeded`, `server_overloaded`, `server_error`, +`resource_not_found`, `request_timeout` and `invalid_request`. Input-policy, +Core/runtime, cancelled and waiting outcomes do not contribute. Successful use +retains the earlier error; comparing timestamps is only a display convention. + +These are best-effort Core receipt times after terminal commit, not provider health +or remote completion times. Private revision identity is independent of timestamps; +old, explicit-provider and historical Sessions cannot update a replacement default. +No public Session/Turn fields or retry identity change. The observation has a +one-second budget including pool/row-lock acquisition and cannot change the committed +Turn. A crash, failure or throttle can omit the last observation indefinitely. + +Errors throttle for 30 seconds regardless of code. Ordinary successful writes +throttle for 30 seconds; the first success after an accepted error records recovery +immediately. For an unchanged revision this allows at most three effective metadata +writes in any half-open 30-second interval of nondecreasing DB-clock time. Equal or +backwards clock readings are preserved and can make timestamp-based display +ambiguous; recovery does not impose a total order. No readiness probe, automatic +refresh, observation history or credential/raw-error read is provided. diff --git a/packages/agents-client/src/admin-client.test.ts b/packages/agents-client/src/admin-client.test.ts index d21b5dc91..712a53263 100644 --- a/packages/agents-client/src/admin-client.test.ts +++ b/packages/agents-client/src/admin-client.test.ts @@ -154,7 +154,7 @@ describe("AdminClient transport boundary", () => { }); it("manages deployment default model providers without ever reading a key", async () => { - const provider = { object: "core.model_provider", harness: "codex", protocol: "responses", base_url: "https://model.example/v1", api_key_configured: true, updated_at: "2026-09-26T08:00:00Z" }; + const provider = { last_used_at: null, last_error_code: null, last_error_at: null, object: "core.model_provider", harness: "codex", protocol: "responses", base_url: "https://model.example/v1", api_key_configured: true, updated_at: "2026-09-26T08:00:00Z" }; const harnesses = { object: "list", data: [ { object: "core.harness", id: "claude_sdk", enabled: false, default: false, model_provider: null }, { object: "core.harness", id: "codex", enabled: true, default: true, model_provider: provider }, diff --git a/packages/agents-client/src/admin-projection.ts b/packages/agents-client/src/admin-projection.ts index 74db035e1..04dcae9f1 100644 --- a/packages/agents-client/src/admin-projection.ts +++ b/packages/agents-client/src/admin-projection.ts @@ -2,7 +2,7 @@ import { AgentCoreError, projectRuntimeObservation, projectSavedAgentConfigurati import { projectTokenUsage } from "./usage-projection"; import { safeProvider } from "./execution-configuration-projection"; import { canonicalUuid, exactFields, isNonnegativeInteger, isRecord, onlyFields, sameResourceId } from "./response-projection"; -import type { CoreHarness, CoreHarnessKind, HarnessModelProvider, ListPage, SavedAgent } from "./types"; +import type { CoreHarness, CoreHarnessKind, HarnessModelProvider, ProviderObservationErrorCode, ListPage, SavedAgent } from "./types"; import type { AdminAPIKey, AdminProject, AdminAuditPage, AdminSummary, AdminRuntimeObservation, RuntimeDiskObservation, AdminKeyProvenance, AdminResourceOwner, AdminWriteOperationPage, AdminAuditResultID, AdminDeleted, AdminIssuedAPIKey, AdminPage, AdminSessionArchive, SessionArtifact, Skill, SkillVersion, ExecutorCredentialList, ExecutorConnection, IssuedExecutorCredential, CoreInstallation, CoreInstallationSetting } from "./admin-types"; export function invalidAdminResponse(): never { @@ -223,16 +223,21 @@ export function projectIssuedExecutorCredential(value: unknown, keyId: string, e return { key_id: issued.key_id, environment_id: issued.environment_id, executor_token: issued.executor_token }; } +const providerObservationErrors = new Set(["authentication_error", "connection_failed", "rate_limit_exceeded", "usage_limit_exceeded", "server_overloaded", "server_error", "resource_not_found", "request_timeout", "invalid_request"]); const harnessKinds = new Set(["claude_sdk", "codex", "mcode"]); /** A deployment default model provider: exactly the safe view, never `api_key`. */ export function projectHarnessModelProvider(value: unknown, harness?: CoreHarnessKind): HarnessModelProvider { - if (!isRecord(value) || !onlyFields(value, new Set(["object", "harness", "updated_at", "protocol", "base_url", "context_window", "max_output_tokens", "api_key_configured"]))) return invalidAdminResponse(); - const { object, harness: kind, updated_at, ...view } = value; + if (!isRecord(value) || !onlyFields(value, new Set(["object", "harness", "updated_at", "last_used_at", "last_error_code", "last_error_at", "protocol", "base_url", "context_window", "max_output_tokens", "api_key_configured"]))) return invalidAdminResponse(); + const { object, harness: kind, updated_at, last_used_at, last_error_code, last_error_at, ...view } = value; + if ((last_used_at !== null && (typeof last_used_at !== "string" || !date(last_used_at))) || + (last_error_at !== null && (typeof last_error_at !== "string" || !date(last_error_at))) || + (last_error_code !== null && (typeof last_error_code !== "string" || !providerObservationErrors.has(last_error_code))) || + ((last_error_code === null) !== (last_error_at === null))) return invalidAdminResponse(); if (object !== "core.model_provider" || typeof kind !== "string" || !harnessKinds.has(kind) || (harness !== undefined && kind !== harness) || typeof updated_at !== "string" || !date(updated_at)) return invalidAdminResponse(); const provider = safeProvider(view, invalidAdminResponse); if (!provider.api_key_configured) return invalidAdminResponse(); - return { object, harness: kind as CoreHarnessKind, ...provider, updated_at }; + return { object, harness: kind as CoreHarnessKind, ...provider, updated_at, last_used_at, last_error_at, last_error_code: last_error_code as ProviderObservationErrorCode | null }; } export function projectCoreHarnessList(value: unknown): { object: "list"; data: CoreHarness[] } { const page = record(value, ["object", "data"]); diff --git a/packages/agents-client/src/provider-observations.test.ts b/packages/agents-client/src/provider-observations.test.ts new file mode 100644 index 000000000..212f82e24 --- /dev/null +++ b/packages/agents-client/src/provider-observations.test.ts @@ -0,0 +1,16 @@ +import { describe, expect, it } from "vitest"; +import { projectHarnessModelProvider } from "./admin-projection"; + +const provider = { object: "core.model_provider", harness: "codex", protocol: "responses", base_url: "https://model.example/v1", api_key_configured: true, updated_at: "2026-09-26T08:00:00Z", last_used_at: null, last_error_code: null, last_error_at: null }; +describe("deployment default observations", () => { + it("preserves null and safe observations", () => { + expect(projectHarnessModelProvider(provider)).toEqual(provider); + const observed = { ...provider, last_used_at: "2026-09-26T08:01:00Z", last_error_at: "2026-09-26T08:00:30Z", last_error_code: "authentication_error" }; + expect(projectHarnessModelProvider(observed)).toEqual(observed); + }); + it("rejects private state, raw codes and inconsistent pairs", () => { + for (const extra of [{ revision: "private" }, { recovery_pending: true }, { last_error_code: "raw provider secret", last_error_at: provider.updated_at }, { last_error_code: "context_length_exceeded", last_error_at: provider.updated_at }, { last_error_code: "server_error" }, { last_error_at: provider.updated_at }, { last_used_at: "invalid" }, { last_used_at: undefined }]) { + expect(() => projectHarnessModelProvider({ ...provider, ...extra })).toThrow(); + } + }); +}); diff --git a/packages/agents-client/src/types.ts b/packages/agents-client/src/types.ts index a074e4ef4..55cf4ddf8 100644 --- a/packages/agents-client/src/types.ts +++ b/packages/agents-client/src/types.ts @@ -1204,7 +1204,12 @@ export interface ModelProviderView { } /** A harness's deployment default model provider in Core. The key is never returned. */ +export type ProviderObservationErrorCode = "authentication_error" | "connection_failed" | "rate_limit_exceeded" | "usage_limit_exceeded" | "server_overloaded" | "server_error" | "resource_not_found" | "request_timeout" | "invalid_request"; + export interface HarnessModelProvider extends ModelProviderView { + last_used_at: string | null; + last_error_code: ProviderObservationErrorCode | null; + last_error_at: string | null; object: "core.model_provider"; harness: CoreHarnessKind; updated_at: string; diff --git a/services/agents-api/internal/api/handler.go b/services/agents-api/internal/api/handler.go index d959075f6..318132d41 100644 --- a/services/agents-api/internal/api/handler.go +++ b/services/agents-api/internal/api/handler.go @@ -245,8 +245,9 @@ func (h *Handler) createSession(w http.ResponseWriter, r *http.Request) { selectedEngine := h.engine var provider *v1.ModelProviderInput var providerSource string + var deploymentRevision uuid.UUID if err == nil { - selectedEngine, provider, providerSource, err = h.resolveSessionExecution(r.Context(), input, inheritedProvider, configuration) + selectedEngine, provider, providerSource, deploymentRevision, err = h.resolveSessionExecution(r.Context(), input, inheritedProvider, configuration) } if err == nil { if invalid := h.policy.ValidateSessionConfiguration(selectedEngine, configuration); invalid != nil { @@ -279,10 +280,11 @@ func (h *Handler) createSession(w http.ResponseWriter, r *http.Request) { } executionConfiguration := sessionExecutionProjection(input, saved, inheritedProvider, provider, selectedEngine, configuration) createInput := store.CreateSessionInput{ - ExecutionConfiguration: &executionConfiguration, - ModelProvider: provider, - ModelProviderSource: providerSource, - Creator: sessionCreator(r), InitialFiles: input.initialFiles, Initialization: input.initialization, + ExecutionConfiguration: &executionConfiguration, + ModelProvider: provider, + ModelProviderSource: providerSource, + DeploymentProviderRevision: deploymentRevision, + Creator: sessionCreator(r), InitialFiles: input.initialFiles, Initialization: input.initialization, Engine: selectedEngine, IdempotencyKey: key, Metadata: input.Metadata, Configuration: configuration, InitialInputs: initialInputs, CreationRequest: creationRequest, } if input.Stream { diff --git a/services/agents-api/internal/api/harness_model_providers.go b/services/agents-api/internal/api/harness_model_providers.go index f912c970b..e516c10d1 100644 --- a/services/agents-api/internal/api/harness_model_providers.go +++ b/services/agents-api/internal/api/harness_model_providers.go @@ -23,14 +23,17 @@ type DeploymentModelProviderStore interface { // HarnessModelProvider is a harness's deployment default model provider. It // never contains the API key, only whether one is configured. type HarnessModelProvider struct { - Object string `json:"object" enums:"core.model_provider" binding:"required"` - Harness string `json:"harness" enums:"claude_sdk,codex,mcode" binding:"required"` - Protocol string `json:"protocol" enums:"anthropic,responses" binding:"required"` - BaseURL string `json:"base_url" binding:"required"` - ContextWindow int32 `json:"context_window,omitempty"` - MaxOutputTokens int32 `json:"max_output_tokens,omitempty"` - APIKeyConfigured bool `json:"api_key_configured" binding:"required"` - UpdatedAt time.Time `json:"updated_at" binding:"required"` + Object string `json:"object" enums:"core.model_provider" binding:"required"` + Harness string `json:"harness" enums:"claude_sdk,codex,mcode" binding:"required"` + Protocol string `json:"protocol" enums:"anthropic,responses" binding:"required"` + BaseURL string `json:"base_url" binding:"required"` + ContextWindow int32 `json:"context_window,omitempty"` + MaxOutputTokens int32 `json:"max_output_tokens,omitempty"` + APIKeyConfigured bool `json:"api_key_configured" binding:"required"` + LastUsedAt *time.Time `json:"last_used_at" format:"date-time" extensions:"x-nullable" binding:"required"` + LastErrorCode *string `json:"last_error_code" extensions:"x-nullable" binding:"required" enums:"authentication_error,connection_failed,rate_limit_exceeded,usage_limit_exceeded,server_overloaded,server_error,resource_not_found,request_timeout,invalid_request"` + LastErrorAt *time.Time `json:"last_error_at" format:"date-time" extensions:"x-nullable" binding:"required"` + UpdatedAt time.Time `json:"updated_at" binding:"required"` } // CoreHarness describes one harness this build supports. Enabled and default @@ -53,7 +56,7 @@ func harnessModelProvider(value store.DeploymentModelProvider) *HarnessModelProv return &HarnessModelProvider{Object: "core.model_provider", Harness: value.Harness, Protocol: value.Provider.Protocol, BaseURL: value.Provider.BaseURL, ContextWindow: value.Provider.ContextWindow, MaxOutputTokens: value.Provider.MaxOutputTokens, - APIKeyConfigured: value.Provider.APIKeyConfigured, UpdatedAt: value.UpdatedAt.UTC()} + APIKeyConfigured: value.Provider.APIKeyConfigured, UpdatedAt: value.UpdatedAt.UTC(), LastUsedAt: value.LastUsedAt, LastErrorCode: value.LastErrorCode, LastErrorAt: value.LastErrorAt} } // registerHarnessRoutes adds harness and deployment model provider management @@ -107,7 +110,7 @@ func (h *Handler) listHarnesses(w http.ResponseWriter, r *http.Request, s Deploy } // @Summary Retrieve a harness's deployment default model provider -// @Description Core key only. Returns the safe view; the key is never returned. 404 when the harness does not exist or has no deployment default. +// @Description Core key only. Returns the safe view; the key is never returned. 404 when the harness does not exist or has no deployment default. Nullable last_used_at, last_error_code and last_error_at are best-effort observations of committed root Turns using this exact default revision; they do not establish current readiness and may remain stale indefinitely. // @Tags Deployment Model Providers // @Produce json // @Security DeploymentAdminAuth @@ -141,7 +144,7 @@ var harnessModelProviderShape = shape{kind: objectValue, members: []member{ }} // @Summary Replace a harness's deployment default model provider -// @Description Core key only. The body is the complete x_agents_core.model_provider bundle, including the write-only api_key; there is no partial update and bundles are never merged. The provider is validated for this harness: an HTTPS base_url without credentials, query or fragment, the harness's protocol (responses for codex, anthropic for claude_sdk and mcode) and, for mcode, positive context_window and max_output_tokens. New openai_hosted and none Sessions that resolve no Session or Agent bundle freeze this default into their encrypted snapshot; existing Sessions never change. self_hosted Sessions never use it. The key is encrypted and never returned. Each write records an administrator audit entry without the key. +// @Description Core key only. The body is the complete x_agents_core.model_provider bundle, including the write-only api_key; there is no partial update and bundles are never merged. The provider is validated for this harness: an HTTPS base_url without credentials, query or fragment, the harness's protocol (responses for codex, anthropic for claude_sdk and mcode) and, for mcode, positive context_window and max_output_tokens. New openai_hosted and none Sessions that resolve no Session or Agent bundle freeze this default into their encrypted snapshot; existing Sessions never change. self_hosted Sessions never use it. The key is encrypted and never returned. Each write records an administrator audit entry without the key and resets last_used_at, last_error_code and last_error_at to null, including identical writes. // @Tags Deployment Model Providers // @Accept json // @Produce json diff --git a/services/agents-api/internal/api/model_provider_fixture_test.go b/services/agents-api/internal/api/model_provider_fixture_test.go index e81fc208b..d2c5eee2d 100644 --- a/services/agents-api/internal/api/model_provider_fixture_test.go +++ b/services/agents-api/internal/api/model_provider_fixture_test.go @@ -2,6 +2,8 @@ package api import ( "context" + "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/store" + "github.com/google/uuid" v1 "github.com/MiniMax-AI-Dev/parsar/contracts/agents-api/v1" ) @@ -19,8 +21,8 @@ func fixtureModelProvider(harness string) *v1.ModelProviderInput { // withFixtureDeploymentProvider configures a deployment default for every harness. func withFixtureDeploymentProvider() Option { - return WithModelProviderDefaults(func(_ context.Context, harness string) (*v1.ModelProviderInput, error) { - return fixtureModelProvider(harness), nil + return WithModelProviderDefaults(func(_ context.Context, harness string) (*store.DeploymentModelProviderSnapshot, error) { + return &store.DeploymentModelProviderSnapshot{Provider: fixtureModelProvider(harness), Revision: uuid.New()}, nil }) } diff --git a/services/agents-api/internal/api/session_model_defaults.go b/services/agents-api/internal/api/session_model_defaults.go index 8e7b574fc..abdede822 100644 --- a/services/agents-api/internal/api/session_model_defaults.go +++ b/services/agents-api/internal/api/session_model_defaults.go @@ -7,12 +7,13 @@ import ( v1 "github.com/MiniMax-AI-Dev/parsar/contracts/agents-api/v1" "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/store" + "github.com/google/uuid" ) // ModelProviderDefaults decrypts the deployment default model provider for a // harness at Session creation, before the encrypted Session snapshot is // committed. It returns nil when the harness has no default. -type ModelProviderDefaults func(context.Context, string) (*v1.ModelProviderInput, error) +type ModelProviderDefaults func(context.Context, string) (*store.DeploymentModelProviderSnapshot, error) func WithModelProviderDefaults(resolve ModelProviderDefaults) Option { return func(h *Handler) { h.modelProviderDefaults = resolve } @@ -74,15 +75,16 @@ func modelProviderRequired(environment, engine string) error { // resolveSessionExecution applies provider precedence: the Session bundle, the // saved Agent bundle, then the deployment default where the environment allows // it. Bundles are never merged. -func (h *Handler) resolveSessionExecution(ctx context.Context, input sessionRequest, inherited *v1.ModelProviderInput, raw json.RawMessage) (string, *v1.ModelProviderInput, string, error) { +func (h *Handler) resolveSessionExecution(ctx context.Context, input sessionRequest, inherited *v1.ModelProviderInput, raw json.RawMessage) (string, *v1.ModelProviderInput, string, uuid.UUID, error) { engine, err := h.sessionHarness(raw) if err != nil { - return "", nil, "", err + return "", nil, "", uuid.Nil, err } + var revision uuid.UUID provider, source := inherited, v1.ModelProviderSourceAgent if extension := input.XAgentsCore; extension != nil { if extension.ModelProvider == nil && !input.modelProviderNull { - return "", nil, "", errors.New("x_agents_core requires an execution option") + return "", nil, "", uuid.Nil, errors.New("x_agents_core requires an execution option") } if extension.ModelProvider != nil { provider, source = extension.ModelProvider, v1.ModelProviderSourceSession @@ -90,23 +92,27 @@ func (h *Handler) resolveSessionExecution(ctx context.Context, input sessionRequ } environment := input.Environment.Type if provider == nil && h.modelProviderDefaults != nil && v1.ModelProviderAllowed(environment, v1.ModelProviderSourceDeployment) { - provider, err = h.modelProviderDefaults(ctx, engine) + var snapshot *store.DeploymentModelProviderSnapshot + snapshot, err = h.modelProviderDefaults(ctx, engine) if err != nil { - return "", nil, "", &modelProviderDefaultsError{err} + return "", nil, "", uuid.Nil, &modelProviderDefaultsError{err} + } + if snapshot != nil { + provider, revision = snapshot.Provider, snapshot.Revision } source = v1.ModelProviderSourceDeployment } if provider == nil { if v1.ModelProviderRequired(environment) { - return "", nil, "", modelProviderRequired(environment, engine) + return "", nil, "", uuid.Nil, modelProviderRequired(environment, engine) } - return engine, nil, "", nil + return engine, nil, "", uuid.Nil, nil } if !v1.ModelProviderAllowed(environment, source) { - return "", nil, "", errors.New("caller model credentials require an openai_hosted or self_hosted environment") + return "", nil, "", uuid.Nil, errors.New("caller model credentials require an openai_hosted or self_hosted environment") } if err := provider.ValidateHarness(engine); err != nil { - return "", nil, "", err + return "", nil, "", uuid.Nil, err } - return engine, provider, source, nil + return engine, provider, source, revision, nil } diff --git a/services/agents-api/internal/db/queries/deployment_model_providers.sql b/services/agents-api/internal/db/queries/deployment_model_providers.sql index bda5a2919..bc53b6eed 100644 --- a/services/agents-api/internal/db/queries/deployment_model_providers.sql +++ b/services/agents-api/internal/db/queries/deployment_model_providers.sql @@ -1,17 +1,18 @@ -- name: ListDeploymentModelProviders :many -SELECT harness, protocol, base_url, context_window, max_output_tokens, updated_at +SELECT harness, protocol, base_url, context_window, max_output_tokens, updated_at, last_used_at, last_error_code, last_error_at FROM deployment_model_providers ORDER BY harness; -- name: GetDeploymentModelProviderSecret :one -SELECT encrypted_config FROM deployment_model_providers WHERE harness = @harness; +SELECT encrypted_config, revision FROM deployment_model_providers WHERE harness = @harness; -- name: UpsertDeploymentModelProvider :one -INSERT INTO deployment_model_providers (harness, protocol, base_url, context_window, max_output_tokens, encrypted_config, updated_at) -VALUES (@harness, @protocol, @base_url, @context_window, @max_output_tokens, @encrypted_config, clock_timestamp()) +INSERT INTO deployment_model_providers (harness, protocol, base_url, context_window, max_output_tokens, encrypted_config, revision, updated_at) +VALUES (@harness, @protocol, @base_url, @context_window, @max_output_tokens, @encrypted_config, @revision, clock_timestamp()) ON CONFLICT (harness) DO UPDATE SET protocol = EXCLUDED.protocol, base_url = EXCLUDED.base_url, context_window = EXCLUDED.context_window, max_output_tokens = EXCLUDED.max_output_tokens, - encrypted_config = EXCLUDED.encrypted_config, updated_at = EXCLUDED.updated_at -RETURNING harness, protocol, base_url, context_window, max_output_tokens, updated_at; + encrypted_config = EXCLUDED.encrypted_config, revision = EXCLUDED.revision, updated_at = EXCLUDED.updated_at, + last_used_at = NULL, last_error_code = NULL, last_error_at = NULL, recovery_pending = false +RETURNING harness, protocol, base_url, context_window, max_output_tokens, updated_at, last_used_at, last_error_code, last_error_at; -- name: DeleteDeploymentModelProvider :execrows DELETE FROM deployment_model_providers WHERE harness = @harness; diff --git a/services/agents-api/internal/db/queries/deployment_provider_observations.sql b/services/agents-api/internal/db/queries/deployment_provider_observations.sql new file mode 100644 index 000000000..e69823bde --- /dev/null +++ b/services/agents-api/internal/db/queries/deployment_provider_observations.sql @@ -0,0 +1,34 @@ +-- name: ObserveDeploymentModelProvider :execrows +-- Root turns live in turns; child turns have their own table. Read only committed +-- outcomes, lock only the matching default, and sample receipt time after its lock. +WITH locked AS MATERIALIZED ( + SELECT d.harness, d.revision, d.last_used_at, d.last_error_at, d.recovery_pending, + t.status = 'completed' AS success, t.outcome->>'engine_error_code' AS provider_code + FROM deployment_model_providers d + JOIN sessions s ON s.engine = d.harness + JOIN session_execution_configuration c ON c.session_id = s.id + JOIN turns t ON t.session_id = s.id + WHERE s.tenant_id = @tenant_id AND s.id = @session_id AND t.id = @turn_id + AND c.configuration #>> '{model_provider,source}' = 'deployment' + AND c.deployment_provider_revision = d.revision + AND (t.status = 'completed' OR + (t.status = 'failed' AND t.outcome->>'error_code' = 'engine_failed' + AND t.outcome->>'engine_error_code' IN ( + 'authentication_error', 'connection_failed', 'rate_limit_exceeded', + 'usage_limit_exceeded', 'server_overloaded', 'server_error', + 'resource_not_found', 'request_timeout', 'invalid_request'))) + FOR UPDATE OF d +), observed AS MATERIALIZED ( + SELECT harness, revision, clock_timestamp() AS received_at FROM locked +) +UPDATE deployment_model_providers d +SET last_used_at = CASE WHEN l.success THEN o.received_at ELSE d.last_used_at END, + last_error_code = CASE WHEN l.success THEN d.last_error_code ELSE l.provider_code END, + last_error_at = CASE WHEN l.success THEN d.last_error_at ELSE o.received_at END, + recovery_pending = NOT l.success +FROM locked l JOIN observed o ON o.harness = l.harness AND o.revision = l.revision +WHERE d.harness = l.harness AND d.revision = l.revision + AND ((l.success AND (l.recovery_pending OR l.last_used_at IS NULL + OR o.received_at >= l.last_used_at + interval '30 seconds')) + OR (NOT l.success AND (l.last_error_at IS NULL + OR o.received_at >= l.last_error_at + interval '30 seconds'))); diff --git a/services/agents-api/internal/db/queries/session_execution_configuration.sql b/services/agents-api/internal/db/queries/session_execution_configuration.sql index c59f9bab6..4d798dab0 100644 --- a/services/agents-api/internal/db/queries/session_execution_configuration.sql +++ b/services/agents-api/internal/db/queries/session_execution_configuration.sql @@ -1,6 +1,6 @@ -- name: SaveSessionExecutionConfiguration :exec -INSERT INTO session_execution_configuration(session_id, configuration) -VALUES (@session_id, @configuration); +INSERT INTO session_execution_configuration(session_id, configuration, deployment_provider_revision) +VALUES (@session_id, @configuration, sqlc.narg(deployment_provider_revision)); -- name: GetSessionExecutionConfiguration :one SELECT s.id, s.engine, s.configuration AS session_configuration, diff --git a/services/agents-api/internal/db/sqlc/deployment_model_providers.sql.go b/services/agents-api/internal/db/sqlc/deployment_model_providers.sql.go index c4ba3687a..fba3e7f95 100644 --- a/services/agents-api/internal/db/sqlc/deployment_model_providers.sql.go +++ b/services/agents-api/internal/db/sqlc/deployment_model_providers.sql.go @@ -24,18 +24,23 @@ func (q *Queries) DeleteDeploymentModelProvider(ctx context.Context, harness str } const getDeploymentModelProviderSecret = `-- name: GetDeploymentModelProviderSecret :one -SELECT encrypted_config FROM deployment_model_providers WHERE harness = $1 +SELECT encrypted_config, revision FROM deployment_model_providers WHERE harness = $1 ` -func (q *Queries) GetDeploymentModelProviderSecret(ctx context.Context, harness string) ([]byte, error) { +type GetDeploymentModelProviderSecretRow struct { + EncryptedConfig []byte `json:"encrypted_config"` + Revision pgtype.UUID `json:"revision"` +} + +func (q *Queries) GetDeploymentModelProviderSecret(ctx context.Context, harness string) (GetDeploymentModelProviderSecretRow, error) { row := q.db.QueryRow(ctx, getDeploymentModelProviderSecret, harness) - var encrypted_config []byte - err := row.Scan(&encrypted_config) - return encrypted_config, err + var i GetDeploymentModelProviderSecretRow + err := row.Scan(&i.EncryptedConfig, &i.Revision) + return i, err } const listDeploymentModelProviders = `-- name: ListDeploymentModelProviders :many -SELECT harness, protocol, base_url, context_window, max_output_tokens, updated_at +SELECT harness, protocol, base_url, context_window, max_output_tokens, updated_at, last_used_at, last_error_code, last_error_at FROM deployment_model_providers ORDER BY harness ` @@ -46,6 +51,9 @@ type ListDeploymentModelProvidersRow struct { ContextWindow int32 `json:"context_window"` MaxOutputTokens int32 `json:"max_output_tokens"` UpdatedAt pgtype.Timestamptz `json:"updated_at"` + LastUsedAt pgtype.Timestamptz `json:"last_used_at"` + LastErrorCode pgtype.Text `json:"last_error_code"` + LastErrorAt pgtype.Timestamptz `json:"last_error_at"` } func (q *Queries) ListDeploymentModelProviders(ctx context.Context) ([]ListDeploymentModelProvidersRow, error) { @@ -64,6 +72,9 @@ func (q *Queries) ListDeploymentModelProviders(ctx context.Context) ([]ListDeplo &i.ContextWindow, &i.MaxOutputTokens, &i.UpdatedAt, + &i.LastUsedAt, + &i.LastErrorCode, + &i.LastErrorAt, ); err != nil { return nil, err } @@ -76,21 +87,23 @@ func (q *Queries) ListDeploymentModelProviders(ctx context.Context) ([]ListDeplo } const upsertDeploymentModelProvider = `-- name: UpsertDeploymentModelProvider :one -INSERT INTO deployment_model_providers (harness, protocol, base_url, context_window, max_output_tokens, encrypted_config, updated_at) -VALUES ($1, $2, $3, $4, $5, $6, clock_timestamp()) +INSERT INTO deployment_model_providers (harness, protocol, base_url, context_window, max_output_tokens, encrypted_config, revision, updated_at) +VALUES ($1, $2, $3, $4, $5, $6, $7, clock_timestamp()) ON CONFLICT (harness) DO UPDATE SET protocol = EXCLUDED.protocol, base_url = EXCLUDED.base_url, context_window = EXCLUDED.context_window, max_output_tokens = EXCLUDED.max_output_tokens, - encrypted_config = EXCLUDED.encrypted_config, updated_at = EXCLUDED.updated_at -RETURNING harness, protocol, base_url, context_window, max_output_tokens, updated_at + encrypted_config = EXCLUDED.encrypted_config, revision = EXCLUDED.revision, updated_at = EXCLUDED.updated_at, + last_used_at = NULL, last_error_code = NULL, last_error_at = NULL, recovery_pending = false +RETURNING harness, protocol, base_url, context_window, max_output_tokens, updated_at, last_used_at, last_error_code, last_error_at ` type UpsertDeploymentModelProviderParams struct { - Harness string `json:"harness"` - Protocol string `json:"protocol"` - BaseUrl string `json:"base_url"` - ContextWindow int32 `json:"context_window"` - MaxOutputTokens int32 `json:"max_output_tokens"` - EncryptedConfig []byte `json:"encrypted_config"` + Harness string `json:"harness"` + Protocol string `json:"protocol"` + BaseUrl string `json:"base_url"` + ContextWindow int32 `json:"context_window"` + MaxOutputTokens int32 `json:"max_output_tokens"` + EncryptedConfig []byte `json:"encrypted_config"` + Revision pgtype.UUID `json:"revision"` } type UpsertDeploymentModelProviderRow struct { @@ -100,6 +113,9 @@ type UpsertDeploymentModelProviderRow struct { ContextWindow int32 `json:"context_window"` MaxOutputTokens int32 `json:"max_output_tokens"` UpdatedAt pgtype.Timestamptz `json:"updated_at"` + LastUsedAt pgtype.Timestamptz `json:"last_used_at"` + LastErrorCode pgtype.Text `json:"last_error_code"` + LastErrorAt pgtype.Timestamptz `json:"last_error_at"` } func (q *Queries) UpsertDeploymentModelProvider(ctx context.Context, arg UpsertDeploymentModelProviderParams) (UpsertDeploymentModelProviderRow, error) { @@ -110,6 +126,7 @@ func (q *Queries) UpsertDeploymentModelProvider(ctx context.Context, arg UpsertD arg.ContextWindow, arg.MaxOutputTokens, arg.EncryptedConfig, + arg.Revision, ) var i UpsertDeploymentModelProviderRow err := row.Scan( @@ -119,6 +136,9 @@ func (q *Queries) UpsertDeploymentModelProvider(ctx context.Context, arg UpsertD &i.ContextWindow, &i.MaxOutputTokens, &i.UpdatedAt, + &i.LastUsedAt, + &i.LastErrorCode, + &i.LastErrorAt, ) return i, err } diff --git a/services/agents-api/internal/db/sqlc/deployment_provider_observations.sql.go b/services/agents-api/internal/db/sqlc/deployment_provider_observations.sql.go new file mode 100644 index 000000000..9f9c0b5a9 --- /dev/null +++ b/services/agents-api/internal/db/sqlc/deployment_provider_observations.sql.go @@ -0,0 +1,62 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.29.0 +// source: deployment_provider_observations.sql + +package sqlc + +import ( + "context" + + "github.com/jackc/pgx/v5/pgtype" +) + +const observeDeploymentModelProvider = `-- name: ObserveDeploymentModelProvider :execrows +WITH locked AS MATERIALIZED ( + SELECT d.harness, d.revision, d.last_used_at, d.last_error_at, d.recovery_pending, + t.status = 'completed' AS success, t.outcome->>'engine_error_code' AS provider_code + FROM deployment_model_providers d + JOIN sessions s ON s.engine = d.harness + JOIN session_execution_configuration c ON c.session_id = s.id + JOIN turns t ON t.session_id = s.id + WHERE s.tenant_id = $1 AND s.id = $2 AND t.id = $3 + AND c.configuration #>> '{model_provider,source}' = 'deployment' + AND c.deployment_provider_revision = d.revision + AND (t.status = 'completed' OR + (t.status = 'failed' AND t.outcome->>'error_code' = 'engine_failed' + AND t.outcome->>'engine_error_code' IN ( + 'authentication_error', 'connection_failed', 'rate_limit_exceeded', + 'usage_limit_exceeded', 'server_overloaded', 'server_error', + 'resource_not_found', 'request_timeout', 'invalid_request'))) + FOR UPDATE OF d +), observed AS MATERIALIZED ( + SELECT harness, revision, clock_timestamp() AS received_at FROM locked +) +UPDATE deployment_model_providers d +SET last_used_at = CASE WHEN l.success THEN o.received_at ELSE d.last_used_at END, + last_error_code = CASE WHEN l.success THEN d.last_error_code ELSE l.provider_code END, + last_error_at = CASE WHEN l.success THEN d.last_error_at ELSE o.received_at END, + recovery_pending = NOT l.success +FROM locked l JOIN observed o ON o.harness = l.harness AND o.revision = l.revision +WHERE d.harness = l.harness AND d.revision = l.revision + AND ((l.success AND (l.recovery_pending OR l.last_used_at IS NULL + OR o.received_at >= l.last_used_at + interval '30 seconds')) + OR (NOT l.success AND (l.last_error_at IS NULL + OR o.received_at >= l.last_error_at + interval '30 seconds'))) +` + +type ObserveDeploymentModelProviderParams struct { + TenantID pgtype.UUID `json:"tenant_id"` + SessionID pgtype.UUID `json:"session_id"` + TurnID pgtype.UUID `json:"turn_id"` +} + +// Root turns live in turns; child turns have their own table. Read only committed +// outcomes, lock only the matching default, and sample receipt time after its lock. +func (q *Queries) ObserveDeploymentModelProvider(ctx context.Context, arg ObserveDeploymentModelProviderParams) (int64, error) { + result, err := q.db.Exec(ctx, observeDeploymentModelProvider, arg.TenantID, arg.SessionID, arg.TurnID) + if err != nil { + return 0, err + } + return result.RowsAffected(), nil +} diff --git a/services/agents-api/internal/db/sqlc/models.go b/services/agents-api/internal/db/sqlc/models.go index f491fa1a7..1760cf046 100644 --- a/services/agents-api/internal/db/sqlc/models.go +++ b/services/agents-api/internal/db/sqlc/models.go @@ -61,6 +61,11 @@ type DeploymentModelProvider struct { MaxOutputTokens int32 `json:"max_output_tokens"` EncryptedConfig []byte `json:"encrypted_config"` UpdatedAt pgtype.Timestamptz `json:"updated_at"` + Revision pgtype.UUID `json:"revision"` + LastUsedAt pgtype.Timestamptz `json:"last_used_at"` + LastErrorCode pgtype.Text `json:"last_error_code"` + LastErrorAt pgtype.Timestamptz `json:"last_error_at"` + RecoveryPending bool `json:"recovery_pending"` } type Device struct { @@ -407,8 +412,9 @@ type SessionEvent struct { } type SessionExecutionConfiguration struct { - SessionID pgtype.UUID `json:"session_id"` - Configuration []byte `json:"configuration"` + SessionID pgtype.UUID `json:"session_id"` + Configuration []byte `json:"configuration"` + DeploymentProviderRevision pgtype.UUID `json:"deployment_provider_revision"` } type SessionItem struct { diff --git a/services/agents-api/internal/db/sqlc/session_execution_configuration.sql.go b/services/agents-api/internal/db/sqlc/session_execution_configuration.sql.go index 9cc3ad926..548bc0226 100644 --- a/services/agents-api/internal/db/sqlc/session_execution_configuration.sql.go +++ b/services/agents-api/internal/db/sqlc/session_execution_configuration.sql.go @@ -44,16 +44,17 @@ func (q *Queries) GetSessionExecutionConfiguration(ctx context.Context, arg GetS } const saveSessionExecutionConfiguration = `-- name: SaveSessionExecutionConfiguration :exec -INSERT INTO session_execution_configuration(session_id, configuration) -VALUES ($1, $2) +INSERT INTO session_execution_configuration(session_id, configuration, deployment_provider_revision) +VALUES ($1, $2, $3) ` type SaveSessionExecutionConfigurationParams struct { - SessionID pgtype.UUID `json:"session_id"` - Configuration []byte `json:"configuration"` + SessionID pgtype.UUID `json:"session_id"` + Configuration []byte `json:"configuration"` + DeploymentProviderRevision pgtype.UUID `json:"deployment_provider_revision"` } func (q *Queries) SaveSessionExecutionConfiguration(ctx context.Context, arg SaveSessionExecutionConfigurationParams) error { - _, err := q.db.Exec(ctx, saveSessionExecutionConfiguration, arg.SessionID, arg.Configuration) + _, err := q.db.Exec(ctx, saveSessionExecutionConfiguration, arg.SessionID, arg.Configuration, arg.DeploymentProviderRevision) return err } diff --git a/services/agents-api/internal/execution/deployment_provider_observations.go b/services/agents-api/internal/execution/deployment_provider_observations.go new file mode 100644 index 000000000..ea2b5c70d --- /dev/null +++ b/services/agents-api/internal/execution/deployment_provider_observations.go @@ -0,0 +1,34 @@ +package execution + +import ( + "context" + "encoding/json" + "time" + + "github.com/MiniMax-AI-Dev/parsar/internal/agentdaemon/proto" + "github.com/MiniMax-AI-Dev/parsar/internal/obs/log" + "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/store" +) + +// Observation is strictly after terminal commit. Its pool/lock timeout cannot +// cancel the execution lease or change the already committed public outcome. +func (d *Dispatcher) observeDeploymentProvider(tenantID, sessionID string, turn store.Turn) { + if turn.Status != store.TurnCompleted { + if turn.Status != store.TurnFailed { + return + } + var result Result + if json.Unmarshal(turn.Outcome, &result) != nil || result.ErrorCode != "engine_failed" { + return + } + code, _ := proto.NormalizeEngineFailure(result.EngineErrorCode, nil) + if code == "" || code == "context_length_exceeded" || code == "cyber_policy" { + return + } + } + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + if _, err := d.Store.ObserveDeploymentModelProvider(ctx, tenantID, sessionID, turn.ID); err != nil { + log.Warn(ctx, "Deployment model provider observation unavailable") + } +} diff --git a/services/agents-api/internal/execution/deployment_provider_observations_test.go b/services/agents-api/internal/execution/deployment_provider_observations_test.go new file mode 100644 index 000000000..9559b9e34 --- /dev/null +++ b/services/agents-api/internal/execution/deployment_provider_observations_test.go @@ -0,0 +1,219 @@ +package execution + +import ( + "context" + "encoding/json" + "errors" + "strings" + "testing" + "time" + + v1 "github.com/MiniMax-AI-Dev/parsar/contracts/agents-api/v1" + "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/adminaudit" + "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/identity" + "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/store" + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgxpool" +) + +type finishObservationFixture struct { + s *store.Store + writer *store.Store + pool *pgxpool.Pool + tenant string + session store.Session + dispatcher Dispatcher +} + +func newFinishObservationFixture(t *testing.T, maxConnections int32) finishObservationFixture { + t.Helper() + var cfg *pgxpool.Config + s, lease := resetManagerStoreConfig(t, func(c *pgxpool.Config) { + if maxConnections > 0 { + c.MaxConns = maxConnections + } + cfg = c.Copy() + }) + pool, err := pgxpool.NewWithConfig(t.Context(), cfg) + if err != nil { + t.Fatal(err) + } + t.Cleanup(pool.Close) + admin := adminaudit.WithSource(t.Context(), adminaudit.Source{CredentialID: "fixture-admin", RequestID: uuid.NewString(), TraceID: uuid.NewString()}) + provider := v1.ModelProviderInput{Protocol: "responses", BaseURL: "https://fixture.example/v1", APIKey: "fixture-only"} + if _, err = s.SetDeploymentModelProvider(admin, "codex", provider); err != nil { + t.Fatal(err) + } + snapshot, err := s.DeploymentModelProvider(t.Context(), "codex") + if err != nil { + t.Fatal(err) + } + tenant := uuid.NewString() + model, harness := "fixture-model", "codex" + input := store.CreateSessionInput{Creator: identity.Subject{Kind: "service_account", ID: "fixture"}, Engine: harness, IdempotencyKey: uuid.NewString(), Configuration: json.RawMessage(`{"agent":{"model":"fixture-model"},"environment":{"type":"none"}}`), ModelProvider: snapshot.Provider, ModelProviderSource: "deployment", DeploymentProviderRevision: snapshot.Revision, + ExecutionConfiguration: &v1.SessionExecutionConfiguration{Model: v1.ExecutionSelection{Value: &model, Source: "session"}, Harness: v1.ExecutionSelection{Value: &harness, Source: "deployment"}, ModelProvider: v1.ExecutionProviderSelection{Source: "deployment"}}} + session, err := s.CreateSession(t.Context(), tenant, input) + if err != nil { + t.Fatal(err) + } + return finishObservationFixture{s, lease.Store(), pool, tenant, session, Dispatcher{Store: lease.Store()}} +} +func (f finishObservationFixture) start(t *testing.T) store.InputReceipt { + t.Helper() + receipt, err := f.s.SubmitMessage(t.Context(), f.tenant, f.session.ID, uuid.NewString(), json.RawMessage(`{"input":[{"role":"user","content":[{"type":"input_text","text":"fixture"}]}]}`)) + if err != nil { + t.Fatal(err) + } + if _, err = f.writer.TransitionTurn(t.Context(), f.tenant, f.session.ID, receipt.TurnID, store.TurnTransition{ExpectedStatus: store.TurnQueued, Status: store.TurnInProgress}); err != nil { + t.Fatal(err) + } + return receipt +} +func (f finishObservationFixture) fields(t *testing.T) (*time.Time, *string) { + t.Helper() + rows, err := f.s.ListDeploymentModelProviders(t.Context()) + if err != nil || len(rows) != 1 { + t.Fatal(err) + } + return rows[0].LastUsedAt, rows[0].LastErrorCode +} +func TestFinishRunObservesOnlyFinalCommittedOutcome(t *testing.T) { + cases := []struct { + name, status, core, native string + unapplied, invalid, used, failed bool + }{ + {name: "success", status: store.TurnCompleted, used: true}, + {name: "provider_failure", status: store.TurnFailed, core: "engine_failed", native: "authentication_error", failed: true}, + {name: "runtime_dominates", status: store.TurnFailed, core: "device_disconnected", native: "authentication_error"}, + {name: "input_policy", status: store.TurnFailed, core: "engine_failed", native: "cyber_policy"}, + {name: "cancelled", status: store.TurnCancelled, core: "engine_failed", native: "authentication_error"}, + {name: "fallback", status: store.TurnCompleted, core: "engine_failed", native: "authentication_error", unapplied: true}, + {name: "invalid_result", status: store.TurnCompleted, invalid: true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + f := newFinishObservationFixture(t, 0) + receipt := f.start(t) + result := Result{ErrorCode: tc.core, EngineErrorCode: tc.native, AppliedThrough: receipt.Sequence} + if tc.unapplied { + result.AppliedThrough = 0 + } + if tc.invalid { + result.Error = strings.Repeat("x", 513*1024) + } + turn, err := f.dispatcher.finishRun(f.tenant, f.session.ID, receipt.TurnID, "fixture-model", result, tc.status) + if err != nil { + t.Fatal(err) + } + if tc.unapplied || tc.invalid { + if turn.Status != store.TurnFailed { + t.Fatal("fallback not final", turn.Status) + } + } + used, code := f.fields(t) + if (used != nil) != tc.used || (code != nil) != tc.failed { + t.Fatalf("unexpected observation: used=%v code=%v", used, code) + } + if _, err = f.pool.Exec(t.Context(), "UPDATE deployment_model_providers SET last_used_at=NULL,last_error_at=NULL,last_error_code=NULL,recovery_pending=false"); err != nil { + t.Fatal(err) + } + _, err = f.dispatcher.finishRun(f.tenant, f.session.ID, receipt.TurnID, "fixture-model", result, tc.status) + if !errors.Is(err, store.ErrTurnConflict) { + t.Fatal("expected completion conflict", err) + } + used, code = f.fields(t) + if used != nil || code != nil { + t.Fatal("failed completion observed") + } + }) + } +} +func TestFinishRunObservationLockTimeoutAndFailureKeepLease(t *testing.T) { + for _, mode := range []string{"lock_timeout", "query_failure", "pool_timeout"} { + t.Run(mode, func(t *testing.T) { + // One leased connection plus one ordinary connection gives an independently + // observable pool-acquisition deadline without starving the completion writer. + max := int32(0) + if mode == "pool_timeout" { + max = 1 + } + f := newFinishObservationFixture(t, 0) + receipt := f.start(t) + var cleanup func() + if mode == "lock_timeout" { + tx, err := f.pool.Begin(t.Context()) + if err != nil { + t.Fatal(err) + } + if _, err = tx.Exec(t.Context(), "SELECT harness FROM deployment_model_providers FOR UPDATE"); err != nil { + t.Fatal(err) + } + cleanup = func() { _ = tx.Rollback(context.Background()) } + } + if mode == "query_failure" { + _, err := f.pool.Exec(t.Context(), `CREATE FUNCTION reject_provider_observation() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN RAISE EXCEPTION 'fixture secret must never be logged'; END $$; CREATE TRIGGER reject_provider_observation BEFORE UPDATE ON deployment_model_providers FOR EACH ROW EXECUTE FUNCTION reject_provider_observation()`) + if err != nil { + t.Fatal(err) + } + } + if mode == "pool_timeout" { + // A separate single-connection Store owns its leased connection. Only the + // best-effort metadata query needs this exhausted pool after completion. + cfg := f.pool.Config().Copy() + cfg.MaxConns = max + pool, err := pgxpool.NewWithConfig(t.Context(), cfg) + if err != nil { + t.Fatal(err) + } + defer pool.Close() + // Observe directly through a Dispatcher using the saturated pool; the final + // commit below remains on the real existing lease. + conn, err := pool.Acquire(t.Context()) + if err != nil { + t.Fatal(err) + } + defer conn.Release() + turn, err := f.writer.CompleteExecution(t.Context(), f.tenant, f.session.ID, receipt.TurnID, store.TurnCompleted, json.RawMessage(`{}`), "", receipt.Sequence) + if err != nil { + t.Fatal(err) + } + d := Dispatcher{Store: store.New(pool)} + started := time.Now() + d.observeDeploymentProvider(f.tenant, f.session.ID, turn) + if elapsed := time.Since(started); elapsed < 900*time.Millisecond || elapsed > 2*time.Second { + t.Fatal("pool timeout exceeded observation budget", elapsed) + } + if err = f.writer.CheckExecutionOwnership(t.Context()); err != nil { + t.Fatal("observation cancelled lease", err) + } + return + } + if cleanup != nil { + defer cleanup() + } + started := time.Now() + turn, err := f.dispatcher.finishRun(f.tenant, f.session.ID, receipt.TurnID, "fixture-model", Result{AppliedThrough: receipt.Sequence}, store.TurnCompleted) + elapsed := time.Since(started) + if err != nil || turn.Status != store.TurnCompleted { + t.Fatal("observation changed commit result", err, turn.Status) + } + if elapsed > 2*time.Second || (mode == "lock_timeout" && elapsed < 900*time.Millisecond) { + t.Fatal("observation duration", elapsed) + } + if cleanup != nil { + cleanup() + } + persisted, err := f.s.GetTurn(t.Context(), f.tenant, f.session.ID, receipt.TurnID) + if err != nil || persisted.Status != store.TurnCompleted { + t.Fatal("terminal outcome lost", err) + } + used, code := f.fields(t) + if used != nil || code != nil { + t.Fatal("failed observation wrote") + } + if err = f.writer.CheckExecutionOwnership(t.Context()); err != nil { + t.Fatal("observation cancelled execution lease", err) + } + }) + } +} diff --git a/services/agents-api/internal/execution/dispatcher.go b/services/agents-api/internal/execution/dispatcher.go index 10aed47a7..83541f628 100644 --- a/services/agents-api/internal/execution/dispatcher.go +++ b/services/agents-api/internal/execution/dispatcher.go @@ -122,7 +122,10 @@ func (d *Dispatcher) finishRun(tenantID, sessionID, turnID, model string, result if errors.Is(err, store.ErrUnappliedInputs) { result.ErrorCode = "input_not_applied" encoded, _ = json.Marshal(result) - return d.Store.CompleteExecution(finishCtx, tenantID, sessionID, turnID, store.TurnFailed, encoded, nativeID, result.AppliedThrough) + turn, err = d.Store.CompleteExecution(finishCtx, tenantID, sessionID, turnID, store.TurnFailed, encoded, nativeID, result.AppliedThrough) + } + if err == nil { + d.observeDeploymentProvider(tenantID, sessionID, turn) } return turn, err } diff --git a/services/agents-api/internal/store/agent_execution_defaults_http_test.go b/services/agents-api/internal/store/agent_execution_defaults_http_test.go index df7a70eb4..f61c62176 100644 --- a/services/agents-api/internal/store/agent_execution_defaults_http_test.go +++ b/services/agents-api/internal/store/agent_execution_defaults_http_test.go @@ -27,10 +27,10 @@ func TestAgentExecutionDefaultsPublicSnapshotAndPrecedence(t *testing.T) { } deployment := &v1.ModelProviderInput{Protocol: "responses", BaseURL: "https://deployment.example/v1", APIKey: "deployment-canary"} defaultsCalls := 0 - handler, err := api.NewHandler(st, auth, "codex", api.WithHarnesses([]string{"codex", "claude_sdk", "mcode"}), api.WithHostedEnvironments(), api.WithExecution(st), api.WithModelProviderDefaults(func(context.Context, string) (*v1.ModelProviderInput, error) { + handler, err := api.NewHandler(st, auth, "codex", api.WithHarnesses([]string{"codex", "claude_sdk", "mcode"}), api.WithHostedEnvironments(), api.WithExecution(st), api.WithModelProviderDefaults(func(context.Context, string) (*store.DeploymentModelProviderSnapshot, error) { defaultsCalls++ copy := *deployment - return ©, nil + return &store.DeploymentModelProviderSnapshot{Provider: ©, Revision: uuid.New()}, nil })) if err != nil { t.Fatal(err) diff --git a/services/agents-api/internal/store/deployment_model_providers.go b/services/agents-api/internal/store/deployment_model_providers.go index 582048fed..a9e7460be 100644 --- a/services/agents-api/internal/store/deployment_model_providers.go +++ b/services/agents-api/internal/store/deployment_model_providers.go @@ -9,7 +9,9 @@ import ( v1 "github.com/MiniMax-AI-Dev/parsar/contracts/agents-api/v1" "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/db/sqlc" + "github.com/google/uuid" "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgtype" ) // ErrModelProviderRequired reports a hosted or self-hosted Session that has no @@ -19,13 +21,20 @@ var ErrModelProviderRequired = errors.New("the Session has no model provider") // DeploymentModelProvider is the safe view of one harness's deployment default. // The API key is write-only; only its presence is reported. type DeploymentModelProvider struct { - Harness string - Provider v1.ModelProviderView - UpdatedAt time.Time + Harness string + Provider v1.ModelProviderView + UpdatedAt time.Time + LastUsedAt *time.Time + LastErrorCode *string + LastErrorAt *time.Time } -func deploymentModelProvider(harness, protocol, baseURL string, contextWindow, maxOutputTokens int32, updatedAt time.Time) DeploymentModelProvider { - return DeploymentModelProvider{Harness: harness, UpdatedAt: updatedAt, Provider: v1.ModelProviderView{ +func deploymentModelProvider(harness, protocol, baseURL string, contextWindow, maxOutputTokens int32, updatedAt time.Time, lastUsedAt, lastErrorAt pgtype.Timestamptz, lastErrorCode pgtype.Text) DeploymentModelProvider { + var code *string + if lastErrorCode.Valid { + code = &lastErrorCode.String + } + return DeploymentModelProvider{Harness: harness, UpdatedAt: updatedAt, LastUsedAt: resetTimestamp(lastUsedAt), LastErrorAt: resetTimestamp(lastErrorAt), LastErrorCode: code, Provider: v1.ModelProviderView{ Protocol: protocol, BaseURL: baseURL, ContextWindow: contextWindow, MaxOutputTokens: maxOutputTokens, APIKeyConfigured: true, }} } @@ -38,7 +47,7 @@ func (s *Store) ListDeploymentModelProviders(ctx context.Context) ([]DeploymentM } result := make([]DeploymentModelProvider, 0, len(rows)) for _, row := range rows { - result = append(result, deploymentModelProvider(row.Harness, row.Protocol, row.BaseUrl, row.ContextWindow, row.MaxOutputTokens, row.UpdatedAt.Time)) + result = append(result, deploymentModelProvider(row.Harness, row.Protocol, row.BaseUrl, row.ContextWindow, row.MaxOutputTokens, row.UpdatedAt.Time, row.LastUsedAt, row.LastErrorAt, row.LastErrorCode)) } return result, nil } @@ -62,12 +71,12 @@ func (s *Store) SetDeploymentModelProvider(ctx context.Context, harness string, q := s.queries.WithTx(tx) row, err := q.UpsertDeploymentModelProvider(ctx, sqlc.UpsertDeploymentModelProviderParams{ Harness: harness, Protocol: provider.Protocol, BaseUrl: provider.BaseURL, - ContextWindow: provider.ContextWindow, MaxOutputTokens: provider.MaxOutputTokens, EncryptedConfig: encrypted, + ContextWindow: provider.ContextWindow, MaxOutputTokens: provider.MaxOutputTokens, EncryptedConfig: encrypted, Revision: pgtype.UUID{Bytes: uuid.New(), Valid: true}, }) if err != nil { return err } - result = deploymentModelProvider(row.Harness, row.Protocol, row.BaseUrl, row.ContextWindow, row.MaxOutputTokens, row.UpdatedAt.Time) + result = deploymentModelProvider(row.Harness, row.Protocol, row.BaseUrl, row.ContextWindow, row.MaxOutputTokens, row.UpdatedAt.Time, row.LastUsedAt, row.LastErrorAt, row.LastErrorCode) return recordDeploymentMutation(ctx, q, "set", "deployment_model_provider", harness) }) return result, err @@ -88,15 +97,15 @@ func (s *Store) DeleteDeploymentModelProvider(ctx context.Context, harness strin // DeploymentModelProvider decrypts a harness's default for Session creation. It // returns nil when none is configured and fails closed when the bundle cannot // be decrypted. -func (s *Store) DeploymentModelProvider(ctx context.Context, harness string) (*v1.ModelProviderInput, error) { - ciphertext, err := s.queries.GetDeploymentModelProviderSecret(ctx, harness) +func (s *Store) DeploymentModelProvider(ctx context.Context, harness string) (*DeploymentModelProviderSnapshot, error) { + snapshot, err := s.queries.GetDeploymentModelProviderSecret(ctx, harness) if errors.Is(err, pgx.ErrNoRows) { return nil, nil } if err != nil { return nil, err } - raw, err := s.credentialCipher.OpenDeploymentModelProvider(ciphertext, harness) + raw, err := s.credentialCipher.OpenDeploymentModelProvider(snapshot.EncryptedConfig, harness) if err != nil { return nil, ErrCredentialStorageUnavailable } @@ -104,5 +113,12 @@ func (s *Store) DeploymentModelProvider(ctx context.Context, harness string) (*v if json.Unmarshal(raw, &provider) != nil || provider.ValidateHarness(harness) != nil { return nil, ErrCredentialStorageUnavailable } - return &provider, nil + return &DeploymentModelProviderSnapshot{Provider: &provider, Revision: uuid.UUID(snapshot.Revision.Bytes)}, nil +} + +// DeploymentModelProviderSnapshot pairs one decrypted bundle with its private +// revision from the same database read. It never enters a public projection. +type DeploymentModelProviderSnapshot struct { + Provider *v1.ModelProviderInput + Revision uuid.UUID `json:"-"` } diff --git a/services/agents-api/internal/store/deployment_model_providers_http_test.go b/services/agents-api/internal/store/deployment_model_providers_http_test.go index c34321d03..45539b22c 100644 --- a/services/agents-api/internal/store/deployment_model_providers_http_test.go +++ b/services/agents-api/internal/store/deployment_model_providers_http_test.go @@ -2,6 +2,7 @@ package store_test import ( "bytes" + "context" "encoding/json" "errors" "net/http/httptest" @@ -112,6 +113,16 @@ func TestDeploymentModelProvidersHTTP(t *testing.T) { } saved := call("PUT", path, coreKey, codexDefault, 200) + for _, field := range []string{"last_used_at", "last_error_code", "last_error_at"} { + if string(saved[field]) != "null" { + t.Fatal("new Core observation field is not null", field) + } + } + for _, field := range []string{"revision", "recovery_pending"} { + if _, ok := saved[field]; ok { + t.Fatal("private observation field exposed", field) + } + } if text(saved["object"]) != "core.model_provider" || text(saved["harness"]) != "codex" || text(saved["base_url"]) != "https://deployment.example/v1" || string(saved["api_key_configured"]) != "true" || text(saved["updated_at"]) == "" { t.Fatalf("unexpected provider view: %v", saved) } @@ -261,6 +272,10 @@ func TestNoneSessionRetryAfterDeploymentDefaultChanges(t *testing.T) { return session.ID } setDefault("first-default-key") + snapshot, err := st.DeploymentModelProvider(t.Context(), "codex") + if err != nil { + t.Fatal(err) + } key := uuid.NewString() original := create(key) if provider, err := st.SessionModelExecution(t.Context(), tenant, original); err != nil || provider.APIKey != "first-default-key" { @@ -279,4 +294,65 @@ func TestNoneSessionRetryAfterDeploymentDefaultChanges(t *testing.T) { if provider, err := st.SessionModelExecution(t.Context(), tenant, original); err != nil || provider.APIKey != "first-default-key" { t.Fatal("retry changed the frozen provider", err) } + var revision uuid.UUID + if err := pool.QueryRow(t.Context(), "SELECT deployment_provider_revision FROM session_execution_configuration WHERE session_id=$1", original).Scan(&revision); err != nil || revision != snapshot.Revision { + t.Fatal("API retry changed frozen revision", err) + } +} + +func TestDeploymentProviderResolutionPairsRevisionDuringReplacement(t *testing.T) { + st, pool := store.NewModelTestStore(t) + tenant, token := uuid.NewString(), uuid.NewString() + auth, err := newTestAuthenticator([]testAPIKey{{OrganizationID: "test-org", ProjectID: uuid.NewString(), SubjectKind: "service_account", SubjectID: "tuple-test", TokenSHA256: device.HashCredential(token), TenantID: tenant}}) + if err != nil { + t.Fatal(err) + } + admin := adminaudit.WithSource(t.Context(), adminaudit.Source{CredentialID: "fixture-admin", RequestID: uuid.NewString(), TraceID: uuid.NewString()}) + provider := v1.ModelProviderInput{Protocol: "responses", BaseURL: "https://original.example/v1", APIKey: "original-fixture-key"} + if _, err = st.SetDeploymentModelProvider(admin, "codex", provider); err != nil { + t.Fatal(err) + } + original, err := st.DeploymentModelProvider(t.Context(), "codex") + if err != nil { + t.Fatal(err) + } + // Replacement after the atomic tuple read but before API resolution returns + // must never pair the old ciphertext with the replacement's revision. + resolver := func(ctx context.Context, harness string) (*store.DeploymentModelProviderSnapshot, error) { + snapshot, err := st.DeploymentModelProvider(ctx, harness) + if err != nil { + return nil, err + } + replacement := provider + replacement.APIKey = "replacement-fixture-key" + _, err = st.SetDeploymentModelProvider(admin, harness, replacement) + return snapshot, err + } + handler, err := api.NewHandler(st, auth, "codex", api.WithExecution(st), api.WithModelProviderDefaults(resolver)) + if err != nil { + t.Fatal(err) + } + r := httptest.NewRequest("POST", "/v1/agents/sessions", strings.NewReader(`{"agent":{"model":"m"},"environment":{"type":"none"},"input":"hello"}`)) + r.Header.Set("Authorization", "Bearer "+token) + r.Header.Set("OpenAI-Beta", "agents=v1") + r.Header.Set("Content-Type", "application/json") + w := httptest.NewRecorder() + handler.ServeHTTP(w, r) + var session struct{ ID string } + if w.Code != 201 || json.Unmarshal(w.Body.Bytes(), &session) != nil { + t.Fatalf("creation failed: %d %s", w.Code, w.Body) + } + var revision uuid.UUID + if err = pool.QueryRow(t.Context(), "SELECT deployment_provider_revision FROM session_execution_configuration WHERE session_id=$1", session.ID).Scan(&revision); err != nil || revision != original.Revision { + t.Fatal("tuple revision changed", err) + } + frozen, err := st.SessionModelExecution(t.Context(), tenant, session.ID) + if err != nil || frozen == nil || *frozen != provider { + t.Fatal("tuple bundle changed", err) + } + for _, private := range []string{"original-fixture-key", "replacement-fixture-key", revision.String(), "deployment_provider_revision"} { + if strings.Contains(w.Body.String(), private) { + t.Fatal("private snapshot entered public response") + } + } } diff --git a/services/agents-api/internal/store/deployment_model_providers_test.go b/services/agents-api/internal/store/deployment_model_providers_test.go index bae4b3662..225b768e6 100644 --- a/services/agents-api/internal/store/deployment_model_providers_test.go +++ b/services/agents-api/internal/store/deployment_model_providers_test.go @@ -45,7 +45,7 @@ func TestDeploymentModelProviderEncryptedAuditedAndReplaced(t *testing.T) { if err != nil || len(listed) != 1 || listed[0].Provider != *provider.SafeView() { t.Fatal("reader without the key could not list safe fields", listed, err) } - if got, err := s.DeploymentModelProvider(ctx, "codex"); err != nil || got == nil || *got != provider { + if got, err := s.DeploymentModelProvider(ctx, "codex"); err != nil || got == nil || *got.Provider != provider { t.Fatal("default did not decrypt", err) } if got, err := s.DeploymentModelProvider(ctx, "mcode"); err != nil || got != nil { @@ -59,7 +59,7 @@ func TestDeploymentModelProviderEncryptedAuditedAndReplaced(t *testing.T) { if _, err := s.SetDeploymentModelProvider(ctx, "codex", replacement); err != nil { t.Fatal(err) } - if got, err := s.DeploymentModelProvider(ctx, "codex"); err != nil || *got != replacement { + if got, err := s.DeploymentModelProvider(ctx, "codex"); err != nil || *got.Provider != replacement { t.Fatal("replacement not complete", err) } deleteCtx := adminaudit.WithSource(t.Context(), adminaudit.Source{CredentialID: "abcd1234", RequestID: "request-delete", TraceID: "trace-delete"}) diff --git a/services/agents-api/internal/store/deployment_provider_observation_lock_test.go b/services/agents-api/internal/store/deployment_provider_observation_lock_test.go new file mode 100644 index 000000000..abd2127fc --- /dev/null +++ b/services/agents-api/internal/store/deployment_provider_observation_lock_test.go @@ -0,0 +1,118 @@ +package store + +import ( + "context" + "testing" + "time" + + "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/db/sqlc" +) + +func TestDeploymentObservationClockSampleFollowsRowLock(t *testing.T) { + f := newProviderObservationFixture(t) + turn := f.terminal(t, TurnCompleted, "", "") + tx, err := f.pool.Begin(t.Context()) + if err != nil { + t.Fatal(err) + } + defer tx.Rollback(context.Background()) + if _, err = tx.Exec(t.Context(), "SELECT harness FROM deployment_model_providers FOR UPDATE"); err != nil { + t.Fatal(err) + } + result := make(chan error, 1) + go func() { + _, err := f.s.ObserveDeploymentModelProvider(t.Context(), f.tenant, f.session.ID, turn.ID) + result <- err + }() + deadline := time.Now().Add(500 * time.Millisecond) + waiting := false + for time.Now().Before(deadline) { + if err = f.pool.QueryRow(t.Context(), "SELECT EXISTS(SELECT 1 FROM pg_stat_activity WHERE datname=current_database() AND wait_event_type='Lock' AND query LIKE '%ObserveDeploymentModelProvider%')").Scan(&waiting); err != nil { + t.Fatal(err) + } + if waiting { + break + } + time.Sleep(time.Millisecond) + } + if !waiting { + t.Fatal("observation did not acquire a lock wait") + } + var beforeUnlock time.Time + if err = tx.QueryRow(t.Context(), "SELECT clock_timestamp()").Scan(&beforeUnlock); err != nil { + t.Fatal(err) + } + if err = tx.Commit(t.Context()); err != nil { + t.Fatal(err) + } + if err = <-result; err != nil { + t.Fatal(err) + } + rows, err := f.s.ListDeploymentModelProviders(t.Context()) + if err != nil { + t.Fatal(err) + } + if rows[0].LastUsedAt == nil || rows[0].LastUsedAt.Before(beforeUnlock) { + t.Fatal("receipt clock sampled before row lock") + } +} + +func TestDeploymentObservationWinningLockIsClearedByPUT(t *testing.T) { + f := newProviderObservationFixture(t) + turn := f.terminal(t, TurnCompleted, "", "") + tx, err := f.pool.Begin(t.Context()) + if err != nil { + t.Fatal(err) + } + defer tx.Rollback(context.Background()) + lookup, _ := turnLookup(f.tenant, f.session.ID, turn.ID) + n, err := sqlc.New(tx).ObserveDeploymentModelProvider(t.Context(), sqlc.ObserveDeploymentModelProviderParams{TenantID: lookup.TenantID, SessionID: lookup.SessionID, TurnID: lookup.ID}) + if err != nil || n != 1 { + t.Fatal("observation did not win row lock", n, err) + } + result := make(chan error, 1) + go func() { + _, err := f.s.SetDeploymentModelProvider(observationAdmin(t), "codex", *f.input.ModelProvider) + result <- err + }() + deadline := time.Now().Add(time.Second) + waiting := false + for time.Now().Before(deadline) { + if err = f.pool.QueryRow(t.Context(), "SELECT EXISTS(SELECT 1 FROM pg_stat_activity WHERE datname=current_database() AND wait_event_type='Lock' AND query LIKE '%UpsertDeploymentModelProvider%')").Scan(&waiting); err != nil { + t.Fatal(err) + } + if waiting { + break + } + time.Sleep(time.Millisecond) + } + if !waiting { + t.Fatal("PUT did not wait for observation") + } + if err = tx.Commit(t.Context()); err != nil { + t.Fatal(err) + } + if err = <-result; err != nil { + t.Fatal(err) + } + rows, err := f.s.ListDeploymentModelProviders(t.Context()) + if err != nil { + t.Fatal(err) + } + if rows[0].LastUsedAt != nil || rows[0].LastErrorCode != nil || rows[0].LastErrorAt != nil { + t.Fatal("PUT retained old revision observation") + } + f.observe(t, turn, 0) +} + +func TestDeploymentObservationWaitingAndNonRootTurnCannotWrite(t *testing.T) { + f := newProviderObservationFixture(t) + receipt := submitMessage(t, f.s, f.tenant, f.session.ID, "waiting") + f.observe(t, Turn{ID: receipt.TurnID}, 0) + transition(t, f.s, f.tenant, f.session.ID, receipt.TurnID, TurnQueued, TurnInProgress) + transition(t, f.s, f.tenant, f.session.ID, receipt.TurnID, TurnInProgress, TurnWaiting) + f.observe(t, Turn{ID: receipt.TurnID}, 0) + // Child Turn identifiers live outside turns. An absent root identifier is + // rejected by the same SQL ownership join, without a child-history lookup. + f.observe(t, Turn{ID: "ffffffff-ffff-4fff-bfff-ffffffffffff"}, 0) +} diff --git a/services/agents-api/internal/store/deployment_provider_observations.go b/services/agents-api/internal/store/deployment_provider_observations.go new file mode 100644 index 000000000..f5a301078 --- /dev/null +++ b/services/agents-api/internal/store/deployment_provider_observations.go @@ -0,0 +1,49 @@ +package store + +import ( + "context" + "strconv" + "time" + + "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/db/sqlc" +) + +// ObserveDeploymentModelProvider attempts one metadata UPDATE through the pool, +// never the leased execution connection. SQL verifies the committed root outcome +// and exact frozen revision without loading credentials. The metadata transaction +// installs server-side timeouts: client cancellation alone can leave PostgreSQL +// executing briefly after pgx has returned a deadline error. +func (s *Store) ObserveDeploymentModelProvider(ctx context.Context, tenantID, sessionID, turnID string) (int64, error) { + lookup, err := turnLookup(tenantID, sessionID, turnID) + if err != nil { + return 0, err + } + ctx, cancel := context.WithTimeout(ctx, time.Second) + defer cancel() + tx, err := s.pool.Begin(ctx) + if err != nil { + return 0, err + } + defer tx.Rollback(ctx) + deadline, _ := ctx.Deadline() + // Leave a small part of the overall budget for returning the server error + // and releasing this metadata-only transaction before the client deadline. + timeout := time.Until(deadline).Milliseconds() - 25 + if timeout <= 0 { + return 0, context.DeadlineExceeded + } + setting := strconv.FormatInt(timeout, 10) + "ms" + if _, err = tx.Exec(ctx, "SELECT set_config('statement_timeout', $1, true), set_config('lock_timeout', $1, true)", setting); err != nil { + return 0, err + } + count, err := sqlc.New(tx).ObserveDeploymentModelProvider(ctx, sqlc.ObserveDeploymentModelProviderParams{ + TenantID: lookup.TenantID, SessionID: lookup.SessionID, TurnID: lookup.ID, + }) + if err != nil { + return 0, err + } + if err = tx.Commit(ctx); err != nil { + return 0, err + } + return count, nil +} diff --git a/services/agents-api/internal/store/deployment_provider_observations_test.go b/services/agents-api/internal/store/deployment_provider_observations_test.go new file mode 100644 index 000000000..8ac698094 --- /dev/null +++ b/services/agents-api/internal/store/deployment_provider_observations_test.go @@ -0,0 +1,424 @@ +package store + +import ( + "context" + "encoding/json" + "fmt" + "os" + "strings" + "sync" + "testing" + "time" + + v1 "github.com/MiniMax-AI-Dev/parsar/contracts/agents-api/v1" + "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/adminaudit" + "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/db/sqlc" + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgtype" + "github.com/jackc/pgx/v5/pgxpool" +) + +type providerObservationFixture struct { + s *Store + pool *pgxpool.Pool + tenant string + input CreateSessionInput + session Session +} + +func observationAdmin(t *testing.T) context.Context { + return adminaudit.WithSource(t.Context(), adminaudit.Source{CredentialID: "fixture-admin", RequestID: uuid.NewString(), TraceID: uuid.NewString()}) +} +func newProviderObservationFixture(t *testing.T) providerObservationFixture { + t.Helper() + s, pool := newManagedTestStore(t) + provider := FixtureModelProvider("codex") + if _, err := s.SetDeploymentModelProvider(observationAdmin(t), "codex", *provider); err != nil { + t.Fatal(err) + } + snapshot, err := s.DeploymentModelProvider(t.Context(), "codex") + if err != nil { + t.Fatal(err) + } + input := executionProjectionInput("deployment") + input.Configuration = json.RawMessage(`{"agent":{"model":"frozen-model"},"environment":{"type":"none"}}`) + input.ModelProvider, input.ModelProviderSource, input.DeploymentProviderRevision = snapshot.Provider, "deployment", snapshot.Revision + input.ExecutionConfiguration.ModelProvider = v1.ExecutionProviderSelection{Source: "deployment"} + tenant := uuid.NewString() + session, err := s.CreateSession(t.Context(), tenant, input) + if err != nil { + t.Fatal(err) + } + return providerObservationFixture{s, pool, tenant, input, session} +} +func (f providerObservationFixture) terminal(t *testing.T, status, coreCode, nativeCode string) Turn { + t.Helper() + receipt := submitMessage(t, f.s, f.tenant, f.session.ID, uuid.NewString()) + transition(t, f.s, f.tenant, f.session.ID, receipt.TurnID, TurnQueued, TurnInProgress) + outcome, _ := json.Marshal(map[string]string{"error_code": coreCode, "engine_error_code": nativeCode}) + turn, err := f.s.CompleteExecution(t.Context(), f.tenant, f.session.ID, receipt.TurnID, status, outcome, "", receipt.Sequence) + if err != nil { + t.Fatal(err) + } + return turn +} +func (f providerObservationFixture) observe(t *testing.T, turn Turn, want int64) { + t.Helper() + n, err := f.s.ObserveDeploymentModelProvider(t.Context(), f.tenant, f.session.ID, turn.ID) + if err != nil || n != want { + t.Fatalf("observation writes=%d want=%d err=%v", n, want, err) + } +} +func (f providerObservationFixture) revision(t *testing.T) uuid.UUID { + t.Helper() + var rev uuid.UUID + if err := f.pool.QueryRow(t.Context(), "SELECT deployment_provider_revision FROM session_execution_configuration WHERE session_id=$1", f.session.ID).Scan(&rev); err != nil { + t.Fatal(err) + } + return rev +} +func TestDeploymentObservationSnapshotReplacementAndRetry(t *testing.T) { + f := newProviderObservationFixture(t) + original := f.revision(t) + // Resolution and insertion have independent boundaries: retain the tuple while + // a PUT replaces the default, then create with that exact earlier tuple. + before, err := f.s.DeploymentModelProvider(t.Context(), "codex") + if err != nil { + t.Fatal(err) + } + replacement := *before.Provider + replacement.APIKey = "different-fixture-key" + if _, err = f.s.SetDeploymentModelProvider(observationAdmin(t), "codex", replacement); err != nil { + t.Fatal(err) + } + if _, err = f.pool.Exec(t.Context(), "UPDATE deployment_model_providers SET updated_at='2000-01-01'"); err != nil { + t.Fatal(err) + } + input := f.input + input.IdempotencyKey = uuid.NewString() + stale, err := f.s.CreateSession(t.Context(), f.tenant, input) + if err != nil { + t.Fatal(err) + } + staleFixture := f + staleFixture.session = stale + if staleFixture.revision(t) != original { + t.Fatal("tuple revision changed during insertion") + } + frozen, err := f.s.SessionModelExecution(t.Context(), f.tenant, stale.ID) + if err != nil || frozen == nil || *frozen != *before.Provider { + t.Fatal("frozen tuple bundle changed", err) + } + staleFixture.observe(t, staleFixture.terminal(t, TurnFailed, "engine_failed", "authentication_error"), 0) + current, _ := f.s.DeploymentModelProvider(t.Context(), "codex") + retry := f.input + retry.ModelProvider = current.Provider + retry.DeploymentProviderRevision = current.Revision + replay, err := f.s.CreateSession(t.Context(), f.tenant, retry) + if err != nil || replay.ID != f.session.ID || f.revision(t) != original { + t.Fatal("retry replaced frozen metadata", err) + } + if err = f.s.DeleteDeploymentModelProvider(observationAdmin(t), "codex"); err != nil { + t.Fatal(err) + } + if _, err = f.s.SetDeploymentModelProvider(observationAdmin(t), "codex", *before.Provider); err != nil { + t.Fatal(err) + } + recreated, _ := f.s.DeploymentModelProvider(t.Context(), "codex") + if recreated.Revision == original || recreated.Revision == current.Revision { + t.Fatal("revision reused") + } + frozenProvider, err := f.s.SessionModelExecution(t.Context(), f.tenant, f.session.ID) + if err != nil || frozenProvider == nil || *frozenProvider != *f.input.ModelProvider { + t.Fatal("migration/replacement changed Session bundle", err) + } + f.observe(t, f.terminal(t, TurnCompleted, "", ""), 0) +} +func TestDeploymentObservationEligibilityAndReset(t *testing.T) { + allowed := []string{"authentication_error", "connection_failed", "rate_limit_exceeded", "usage_limit_exceeded", "server_overloaded", "server_error", "resource_not_found", "request_timeout", "invalid_request"} + f := newProviderObservationFixture(t) + for _, code := range allowed { + if _, err := f.pool.Exec(t.Context(), "UPDATE deployment_model_providers SET last_error_at=NULL,last_error_code=NULL,recovery_pending=false"); err != nil { + t.Fatal(err) + } + f.observe(t, f.terminal(t, TurnFailed, "engine_failed", code), 1) + } + for _, tc := range []struct{ status, core, code string }{{TurnFailed, "engine_failed", "context_length_exceeded"}, {TurnFailed, "engine_failed", "cyber_policy"}, {TurnFailed, "engine_failed", "harness_error"}, {TurnFailed, "engine_failed", "untrusted raw text"}, {TurnFailed, "input_not_applied", "authentication_error"}, {TurnFailed, "device_disconnected", "authentication_error"}, {TurnCancelled, "engine_failed", "authentication_error"}} { + f.observe(t, f.terminal(t, tc.status, tc.core, tc.code), 0) + } + success := f.terminal(t, TurnCompleted, "", "") + f.observe(t, success, 1) + n, err := f.s.ObserveDeploymentModelProvider(t.Context(), uuid.NewString(), f.session.ID, success.ID) + if n != 0 || err != nil { + t.Fatal("foreign tenant observed", err) + } + view, _ := f.s.ListDeploymentModelProviders(t.Context()) + if view[0].LastUsedAt == nil || view[0].LastErrorAt == nil { + t.Fatal("safe observations missing") + } + old := f.revision(t) + reset, err := f.s.SetDeploymentModelProvider(observationAdmin(t), "codex", *f.input.ModelProvider) + if err != nil { + t.Fatal(err) + } + snapshot, _ := f.s.DeploymentModelProvider(t.Context(), "codex") + if snapshot.Revision == old || reset.LastUsedAt != nil || reset.LastErrorAt != nil || reset.LastErrorCode != nil { + t.Fatal("identical PUT did not reset") + } + f.observe(t, success, 0) +} +func TestDeploymentObservationSourceAndHistoricalExclusion(t *testing.T) { + f := newProviderObservationFixture(t) + for _, source := range []string{"session", "agent", "unknown", "deployment"} { + input := f.input + input.IdempotencyKey = uuid.NewString() + projection := *input.ExecutionConfiguration + input.ExecutionConfiguration = &projection + input.ModelProviderSource = source + // Historical metadata may name deployment but has no frozen private UUID. + if source == "deployment" { + input.DeploymentProviderRevision = uuid.Nil + } else { + input.Configuration = json.RawMessage(`{"agent":{"model":"frozen-model"},"environment":{"type":"openai_hosted"}}`) + if source == "unknown" { + input.ModelProviderSource = "session" + } + } + projection.ModelProvider = v1.ExecutionProviderSelection{Source: source, Status: "available", Configuration: input.ModelProvider.SafeView()} + session, err := f.s.CreateSession(t.Context(), f.tenant, input) + if err != nil { + t.Fatal(source, err) + } + var revision pgtype.UUID + if err = f.pool.QueryRow(t.Context(), "SELECT deployment_provider_revision FROM session_execution_configuration WHERE session_id=$1", session.ID).Scan(&revision); err != nil || revision.Valid { + t.Fatal("non-deployment/historical revision persisted", source, err) + } + // Use a committed fixture outcome directly: hosted dispatch is a separate gate. + id := uuid.NewString() + if _, err = f.pool.Exec(t.Context(), "INSERT INTO turns(id,session_id,status,outcome,completed_at) VALUES($1,$2,'completed','{}',clock_timestamp())", id, session.ID); err != nil { + t.Fatal(err) + } + n, err := f.s.ObserveDeploymentModelProvider(t.Context(), f.tenant, session.ID, id) + if err != nil || n != 0 { + t.Fatal("ineligible source observed", source, err) + } + } +} +func TestDeploymentObservationConcurrentThrottleAndRecovery(t *testing.T) { + f := newProviderObservationFixture(t) + successes := []Turn{} + failures := []Turn{} + for range 8 { + successes = append(successes, f.terminal(t, TurnCompleted, "", "")) + } + for i := range 8 { + code := []string{"authentication_error", "rate_limit_exceeded"}[i%2] + failures = append(failures, f.terminal(t, TurnFailed, "engine_failed", code)) + } + concurrent := func(turns []Turn, want int64) { + t.Helper() + var wg sync.WaitGroup + counts := make(chan int64, len(turns)) + errs := make(chan error, len(turns)) + for _, turn := range turns { + wg.Add(1) + go func() { + defer wg.Done() + n, err := f.s.ObserveDeploymentModelProvider(t.Context(), f.tenant, f.session.ID, turn.ID) + counts <- n + errs <- err + }() + } + wg.Wait() + close(counts) + close(errs) + var n int64 + for v := range counts { + n += v + } + for err := range errs { + if err != nil { + t.Fatal(err) + } + } + if n != want { + t.Fatalf("concurrent writes=%d want=%d", n, want) + } + } + concurrent(successes, 1) + concurrent(failures, 1) + concurrent(successes, 1) + concurrent(successes, 0) + concurrent(failures, 0) +} + +// Only the DB clock sample is controlled; execute the actual generated query, +// retaining PostgreSQL locking, predicates, constraints and write count. +type observationClockDB struct { + *pgxpool.Pool + at time.Time + query string +} + +func (db *observationClockDB) Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error) { + db.query = sql + sql = strings.ReplaceAll(sql, "clock_timestamp()", "'"+db.at.Format(time.RFC3339Nano)+"'::timestamptz") + return db.Pool.Exec(ctx, sql, args...) +} +func TestDeploymentObservationClockBoundariesRollbackAndPlan(t *testing.T) { + f := newProviderObservationFixture(t) + success := f.terminal(t, TurnCompleted, "", "") + failure := f.terminal(t, TurnFailed, "engine_failed", "authentication_error") + base := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + db := &observationClockDB{Pool: f.pool} + q := sqlc.New(db) + observe := func(turn Turn, offset time.Duration, want int64) { + t.Helper() + db.at = base.Add(offset) + lookup, _ := turnLookup(f.tenant, f.session.ID, turn.ID) + n, err := q.ObserveDeploymentModelProvider(t.Context(), sqlc.ObserveDeploymentModelProviderParams{TenantID: lookup.TenantID, SessionID: lookup.SessionID, TurnID: lookup.ID}) + if err != nil || n != want { + t.Fatalf("at %s got %d want %d: %v", offset, n, want, err) + } + } + observe(success, 0, 1) + observe(failure, time.Second, 1) + observe(success, 2*time.Second, 1) + observe(failure, 30*time.Second, 0) + observe(failure, 31*time.Second, 1) + observe(success, 31*time.Second, 1) + observe(success, 31*time.Second, 0) + // A new accepted error can recover once even if the clock moves backwards. + observe(failure, 61*time.Second, 1) + observe(success, -time.Second, 1) + observe(success, -time.Second, 0) + observe(success, 29*time.Second, 1) + observe(success, 59*time.Second-time.Microsecond, 0) + observe(success, 59*time.Second, 1) + var actual time.Time + if err := f.pool.QueryRow(t.Context(), "SELECT last_used_at FROM deployment_model_providers").Scan(&actual); err != nil || !actual.Equal(base.Add(59*time.Second)) { + t.Fatal("timestamp was clamped", err) + } + lookup, _ := turnLookup(f.tenant, f.session.ID, success.ID) + // EXPLAIN uses the normal production planner and does not execute the update. + // Tenant/Session and root Turn lookups must constrain the default lookup. + tx, err := f.pool.Begin(t.Context()) + if err != nil { + t.Fatal(err) + } + defer tx.Rollback(context.Background()) + rows, err := tx.Query(t.Context(), "EXPLAIN "+db.query, lookup.TenantID, lookup.SessionID, lookup.ID) + if err != nil { + t.Fatal(err) + } + var plan strings.Builder + for rows.Next() { + var line string + _ = rows.Scan(&line) + plan.WriteString(line) + } + rows.Close() + if rows.Err() != nil || (!strings.Contains(plan.String(), "deployment_model_providers_pkey") || (!strings.Contains(plan.String(), "turns_pkey") && !strings.Contains(plan.String(), "turns_session_id_id_key"))) { + t.Fatal("bounded lookup indexes absent", rows.Err(), plan.String()) + } +} +func TestDeploymentObservationReplacementLockRecheckAndTimeout(t *testing.T) { + for _, remove := range []bool{false, true} { + t.Run(fmt.Sprint("delete=", remove), func(t *testing.T) { + f := newProviderObservationFixture(t) + turn := f.terminal(t, TurnCompleted, "", "") + tx, err := f.pool.Begin(t.Context()) + if err != nil { + t.Fatal(err) + } + defer tx.Rollback(context.Background()) + if _, err = tx.Exec(t.Context(), "SELECT harness FROM deployment_model_providers FOR UPDATE"); err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(t.Context(), 80*time.Millisecond) + defer cancel() + if n, err := f.s.ObserveDeploymentModelProvider(ctx, f.tenant, f.session.ID, turn.ID); err == nil || n != 0 { + t.Fatal("locked observation did not time out") + } + done := make(chan int64, 1) + errs := make(chan error, 1) + go func() { + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + n, err := f.s.ObserveDeploymentModelProvider(ctx, f.tenant, f.session.ID, turn.ID) + done <- n + errs <- err + }() + // Verify it actually reached a PostgreSQL lock wait, rather than racing on + // goroutine scheduling, before replacing/deleting the selected revision. + deadline := time.Now().Add(time.Second) + waiting := false + for time.Now().Before(deadline) { + if err = f.pool.QueryRow(t.Context(), "SELECT EXISTS(SELECT 1 FROM pg_stat_activity WHERE datname=current_database() AND wait_event_type='Lock' AND query LIKE '%ObserveDeploymentModelProvider%')").Scan(&waiting); err != nil { + t.Fatal(err) + } + if waiting { + break + } + time.Sleep(time.Millisecond) + } + if !waiting { + t.Fatal("observation never waited on row lock") + } + if remove { + _, err = tx.Exec(t.Context(), "DELETE FROM deployment_model_providers") + } else { + _, err = tx.Exec(t.Context(), "UPDATE deployment_model_providers SET revision=$1", uuid.New()) + } + if err != nil { + t.Fatal(err) + } + if err = tx.Commit(t.Context()); err != nil { + t.Fatal(err) + } + if n := <-done; n != 0 { + t.Fatal("stale revision updated replacement", n) + } + if err = <-errs; err != nil { + t.Fatal(err) + } + }) + } +} +func TestDeploymentObservationMigrationRoundTrip(t *testing.T) { + f := newProviderObservationFixture(t) + old := f.revision(t) + var secret []byte + if err := f.pool.QueryRow(t.Context(), "SELECT encrypted_config FROM deployment_model_providers").Scan(&secret); err != nil { + t.Fatal(err) + } + raw, err := os.ReadFile("../../migrations/000084_deployment_provider_observations.sql") + if err != nil { + t.Fatal(err) + } + parts := strings.Split(string(raw), "-- +goose Down") + if _, err = f.pool.Exec(t.Context(), parts[1]); err != nil { + t.Fatal(err) + } + if _, err = f.pool.Exec(t.Context(), parts[0]); err != nil { + t.Fatal(err) + } + var revision uuid.UUID + var frozen pgtype.UUID + var preserved []byte + if err = f.pool.QueryRow(t.Context(), "SELECT revision,encrypted_config FROM deployment_model_providers").Scan(&revision, &preserved); err != nil { + t.Fatal(err) + } + if err = f.pool.QueryRow(t.Context(), "SELECT deployment_provider_revision FROM session_execution_configuration WHERE session_id=$1", f.session.ID).Scan(&frozen); err != nil { + t.Fatal(err) + } + if revision == old || frozen.Valid || string(secret) != string(preserved) { + t.Fatal("migration recreated historical identity or changed ciphertext") + } + frozenProvider, err := f.s.SessionModelExecution(t.Context(), f.tenant, f.session.ID) + if err != nil || frozenProvider == nil || *frozenProvider != *f.input.ModelProvider { + t.Fatal("migration/replacement changed Session bundle", err) + } + f.observe(t, f.terminal(t, TurnCompleted, "", ""), 0) +} diff --git a/services/agents-api/internal/store/model_provider_fixture_test.go b/services/agents-api/internal/store/model_provider_fixture_test.go index e19d0e2aa..7657839a0 100644 --- a/services/agents-api/internal/store/model_provider_fixture_test.go +++ b/services/agents-api/internal/store/model_provider_fixture_test.go @@ -3,8 +3,8 @@ package store_test import ( "context" "encoding/json" + "github.com/google/uuid" - v1 "github.com/MiniMax-AI-Dev/parsar/contracts/agents-api/v1" "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/api" "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/store" ) @@ -15,8 +15,8 @@ import ( // fixtureDeploymentProvider configures a deployment default for every harness. func fixtureDeploymentProvider() api.Option { - return api.WithModelProviderDefaults(func(_ context.Context, harness string) (*v1.ModelProviderInput, error) { - return store.FixtureModelProvider(harness), nil + return api.WithModelProviderDefaults(func(_ context.Context, harness string) (*store.DeploymentModelProviderSnapshot, error) { + return &store.DeploymentModelProviderSnapshot{Provider: store.FixtureModelProvider(harness), Revision: uuid.New()}, nil }) } diff --git a/services/agents-api/internal/store/session_execution_configuration.go b/services/agents-api/internal/store/session_execution_configuration.go index 593987609..14b8ef35b 100644 --- a/services/agents-api/internal/store/session_execution_configuration.go +++ b/services/agents-api/internal/store/session_execution_configuration.go @@ -10,14 +10,16 @@ import ( "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/db/sqlc" "github.com/google/uuid" "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgtype" ) // The projection is creation metadata, excluded from retry identity. Only the // transaction that inserts the Session writes it; retries cannot repair or alter it. -func saveSessionExecutionConfiguration(ctx context.Context, q *sqlc.Queries, session sqlc.Session, projection *v1.SessionExecutionConfiguration, provider *v1.ModelProviderInput) error { +func saveSessionExecutionConfiguration(ctx context.Context, q *sqlc.Queries, session sqlc.Session, projection *v1.SessionExecutionConfiguration, provider *v1.ModelProviderInput, providerSource string, revision uuid.UUID) error { if projection == nil { return nil } + var frozenRevision pgtype.UUID frozen := *projection model, err := sessionExecutionModel(session.Configuration) if err != nil { @@ -37,6 +39,9 @@ func saveSessionExecutionConfiguration(ctx context.Context, q *sqlc.Queries, ses // are never part of it. frozen.ModelProvider.Status = "available" frozen.ModelProvider.Configuration = provider.SafeView() + if providerSource == v1.ModelProviderSourceDeployment && revision != uuid.Nil { + frozenRevision = pgtype.UUID{Bytes: revision, Valid: true} + } case "session", "agent": if provider == nil || frozen.ModelProvider.Status != "available" || frozen.ModelProvider.Configuration == nil || *frozen.ModelProvider.Configuration != *provider.SafeView() { return fmt.Errorf("%w: execution projection does not match model provider", ErrInvalidInput) @@ -52,7 +57,7 @@ func saveSessionExecutionConfiguration(ctx context.Context, q *sqlc.Queries, ses if err != nil { return err } - return q.SaveSessionExecutionConfiguration(ctx, sqlc.SaveSessionExecutionConfigurationParams{SessionID: session.ID, Configuration: raw}) + return q.SaveSessionExecutionConfiguration(ctx, sqlc.SaveSessionExecutionConfigurationParams{SessionID: session.ID, Configuration: raw, DeploymentProviderRevision: frozenRevision}) } // GetSessionExecutionConfiguration reads only safe committed configuration. It diff --git a/services/agents-api/internal/store/session_initial_input.go b/services/agents-api/internal/store/session_initial_input.go index c8566e006..ecad2ba78 100644 --- a/services/agents-api/internal/store/session_initial_input.go +++ b/services/agents-api/internal/store/session_initial_input.go @@ -23,7 +23,7 @@ func validateInitialInputs(inputs []Input) ([]Input, json.RawMessage, error) { // The Session upsert locks retries. Only the new row reserves or admits work, so a // retry after completion or later Turns cannot submit the original input again. -func (s *Store) createSessionResources(ctx context.Context, tenant string, params sqlc.CreateSessionParams, inputs []Input, encodedInput json.RawMessage, files []InitialFile, setup EnvironmentSetup, provider *v1.ModelProviderInput, executionConfiguration *v1.SessionExecutionConfiguration) (sqlc.Session, *Environment, error) { +func (s *Store) createSessionResources(ctx context.Context, tenant string, params sqlc.CreateSessionParams, inputs []Input, encodedInput json.RawMessage, files []InitialFile, setup EnvironmentSetup, provider *v1.ModelProviderInput, executionConfiguration *v1.SessionExecutionConfiguration, providerSource string, deploymentRevision uuid.UUID) (sqlc.Session, *Environment, error) { var row sqlc.Session var environment *Environment err := pgx.BeginFunc(ctx, s.pool, func(tx pgx.Tx) error { @@ -54,7 +54,7 @@ func (s *Store) createSessionResources(ctx context.Context, tenant string, param if err := s.saveSessionModelExecution(ctx, q, tenant, row.ID, provider); err != nil { return err } - if err := saveSessionExecutionConfiguration(ctx, q, row, executionConfiguration, provider); err != nil { + if err := saveSessionExecutionConfiguration(ctx, q, row, executionConfiguration, provider, providerSource, deploymentRevision); err != nil { return err } if len(files) > 0 { diff --git a/services/agents-api/internal/store/sessions.go b/services/agents-api/internal/store/sessions.go index a25c0b35e..189a6be47 100644 --- a/services/agents-api/internal/store/sessions.go +++ b/services/agents-api/internal/store/sessions.go @@ -57,18 +57,20 @@ type Session struct { } type CreateSessionInput struct { - ExecutionConfiguration *v1.SessionExecutionConfiguration - ModelProvider *v1.ModelProviderInput - ModelProviderSource string // session, agent or deployment; empty allows only openai_hosted - Initialization EnvironmentSetup - InitialFiles []InitialFile - Creator identity.Subject - CreationRequest json.RawMessage - Engine string - Metadata map[string]string - IdempotencyKey string - Configuration json.RawMessage - InitialInputs []Input + // DeploymentProviderRevision is private creation metadata, never retry identity. + DeploymentProviderRevision uuid.UUID `json:"-"` + ExecutionConfiguration *v1.SessionExecutionConfiguration + ModelProvider *v1.ModelProviderInput + ModelProviderSource string // session, agent or deployment; empty allows only openai_hosted + Initialization EnvironmentSetup + InitialFiles []InitialFile + Creator identity.Subject + CreationRequest json.RawMessage + Engine string + Metadata map[string]string + IdempotencyKey string + Configuration json.RawMessage + InitialInputs []Input } type SessionPage struct { @@ -193,7 +195,7 @@ func (s *Store) createSession(ctx context.Context, tenantID string, input Create Configuration: configuration, CreationRequestHash: creationHash, CreatorKind: pgtype.Text{String: input.Creator.Kind, Valid: true}, CreatorID: pgtype.Text{String: input.Creator.ID, Valid: true}, } - row, environment, err := s.createSessionResources(ctx, tenantID, params, batch, encodedInput, input.InitialFiles, input.Initialization, input.ModelProvider, input.ExecutionConfiguration) + row, environment, err := s.createSessionResources(ctx, tenantID, params, batch, encodedInput, input.InitialFiles, input.Initialization, input.ModelProvider, input.ExecutionConfiguration, input.ModelProviderSource, input.DeploymentProviderRevision) if errors.Is(err, pgx.ErrNoRows) { return SessionCreation{}, ErrIdempotencyConflict } diff --git a/services/agents-api/migrations/000084_deployment_provider_observations.sql b/services/agents-api/migrations/000084_deployment_provider_observations.sql new file mode 100644 index 000000000..301a8ef33 --- /dev/null +++ b/services/agents-api/migrations/000084_deployment_provider_observations.sql @@ -0,0 +1,27 @@ +-- +goose Up +-- Revision identity is private and independent of content and wall-clock time. +ALTER TABLE deployment_model_providers + ADD COLUMN revision uuid NOT NULL DEFAULT gen_random_uuid(), + ADD COLUMN last_used_at timestamptz, + ADD COLUMN last_error_code text, + ADD COLUMN last_error_at timestamptz, + ADD COLUMN recovery_pending boolean NOT NULL DEFAULT false, + ADD CONSTRAINT deployment_provider_error_pair CHECK ((last_error_code IS NULL) = (last_error_at IS NULL)), + ADD CONSTRAINT deployment_provider_recovery CHECK (NOT recovery_pending OR last_error_at IS NOT NULL), + ADD CONSTRAINT deployment_provider_error_code CHECK (last_error_code IN ( + 'authentication_error', 'connection_failed', 'rate_limit_exceeded', + 'usage_limit_exceeded', 'server_overloaded', 'server_error', + 'resource_not_found', 'request_timeout', 'invalid_request')); +ALTER TABLE session_execution_configuration ADD COLUMN deployment_provider_revision uuid; + +-- +goose Down +ALTER TABLE session_execution_configuration DROP COLUMN deployment_provider_revision; +ALTER TABLE deployment_model_providers + DROP CONSTRAINT deployment_provider_error_pair, + DROP CONSTRAINT deployment_provider_recovery, + DROP CONSTRAINT deployment_provider_error_code, + DROP COLUMN revision, + DROP COLUMN last_used_at, + DROP COLUMN last_error_code, + DROP COLUMN last_error_at, + DROP COLUMN recovery_pending; From d990c38989650a29a3e451e7b8e35a234fbbbd55 Mon Sep 17 00:00:00 2001 From: SaladDay <1203511142@qq.com> Date: Mon, 28 Sep 2026 09:30:48 +0000 Subject: [PATCH 2/5] fix(client): preserve exact safe E2B error parameters --- contracts/agents-api/core-errors.md | 7 +- packages/agents-client/src/sandbox-client.ts | 10 ++- .../src/sandbox-e2b-error-params.test.ts | 78 +++++++++++++++++++ 3 files changed, 91 insertions(+), 4 deletions(-) create mode 100644 packages/agents-client/src/sandbox-e2b-error-params.test.ts diff --git a/contracts/agents-api/core-errors.md b/contracts/agents-api/core-errors.md index d895cc245..827efd01b 100644 --- a/contracts/agents-api/core-errors.md +++ b/contracts/agents-api/core-errors.md @@ -73,7 +73,12 @@ message, template name, key or unlisted-resource count is returned in details. Missing or unsettled Create receipts are uncertainty, never evidence of a different team or released compute. The typed client projects these codes to fixed local -messages. It also projects `409 sandbox_configuration_error` to fixed public-URL +messages. It preserves the three nonnull fields above only when the response +status, code and param exactly match the table; all other params on these fixed +errors become null. Numeric details remain limited to `current_generation` for +`sandbox_generation_stale` and `allocations`/`pending` for `sandbox_in_use`. Unknown +credential-bearing errors become `sandbox_configuration_unconfirmed` without replay. +It also projects `409 sandbox_configuration_error` to fixed public-URL guidance, even when a PUT omitted its key; arbitrary upstream text is never echoed. ## Operation validation diff --git a/packages/agents-client/src/sandbox-client.ts b/packages/agents-client/src/sandbox-client.ts index fa1e72868..7118571a9 100644 --- a/packages/agents-client/src/sandbox-client.ts +++ b/packages/agents-client/src/sandbox-client.ts @@ -352,12 +352,16 @@ export class SandboxAdminClient { e2b_template_build_invalid: "Select a ready immutable E2B template build with matching resources.", e2b_request_unconfirmed: "E2B verification could not be confirmed. Refresh before submitting again.", }; - if (error.code && messages[error.code]) { + if (error.code && Object.hasOwn(messages, error.code)) { // Credential-bearing errors expose fixed local copy and allowlisted - // numeric facts only; never echo arbitrary message, param or details. + // numeric facts plus exact status/code/field matches only. const fields = error.code === "sandbox_generation_stale" ? ["current_generation"] : error.code === "sandbox_in_use" ? ["allocations", "pending"] : []; const details = Object.fromEntries(fields.filter(field => isNonnegativeInteger(error.details?.[field])).map(field => [field, Number(error.details![field])])); - throw new AgentCoreError(messages[error.code]!, error.status, error.code, null, undefined, Object.keys(details).length ? details : undefined); + const safeParam = error.status === 400 + ? error.code === "e2b_api_key_invalid" ? "e2b.api_key" : error.code === "e2b_template_build_invalid" ? "e2b.template" : null + : error.status === 409 && error.code === "e2b_team_mismatch" ? "e2b.api_key" : null; + const param = error.param === safeParam ? safeParam : null; + throw new AgentCoreError(messages[error.code]!, error.status, error.code, param, undefined, Object.keys(details).length ? details : undefined); } } // This code has one fixed Core meaning. Never forward its raw message or diff --git a/packages/agents-client/src/sandbox-e2b-error-params.test.ts b/packages/agents-client/src/sandbox-e2b-error-params.test.ts new file mode 100644 index 000000000..a62db9b69 --- /dev/null +++ b/packages/agents-client/src/sandbox-e2b-error-params.test.ts @@ -0,0 +1,78 @@ +import { describe, expect, it, vi } from "vitest"; +import { AgentCoreError } from "./client"; +import { SandboxAdminClient } from "./sandbox-client"; + +const cases = [ + { status: 400, code: "e2b_api_key_invalid", param: "e2b.api_key", message: "The E2B API key was rejected." }, + { status: 400, code: "e2b_template_build_invalid", param: "e2b.template", message: "Select a ready immutable E2B template build with matching resources." }, + { status: 409, code: "e2b_team_mismatch", param: "e2b.api_key", message: "This E2B key cannot manage the retained deployment. Reset before changing teams." }, +] as const; +const currentKey = "current-secret-canary"; +const storedKey = "stored-secret-canary"; +const reflected = `${currentKey} ${storedKey}`; + +type Rejection = { status: number; code: string; param?: unknown; details?: unknown }; +async function reject(input: Rejection, method: "POST" | "PUT" = "PUT", includeKey = true) { + const fetch = vi.fn().mockResolvedValue(new Response(JSON.stringify({ error: { + ...input, message: reflected, type: reflected, + } }), { status: input.status, headers: { "Content-Type": "application/json" } })); + const client = new SandboxAdminClient({ fetch }); + const e2b = { template: "runtime:00000000-0000-0000-0000-000000000001", ...(includeKey ? { api_key: currentKey } : {}) }; + const result = await (method === "POST" + ? client.initializeDeployment({ provider: "e2b", expected_generation: 0, e2b: { ...e2b, api_key: currentKey } }) + : client.updateDeployment({ provider: "e2b", expected_generation: 1, e2b })).catch((error: unknown) => error); + expect(result).toBeInstanceOf(AgentCoreError); + const error = result as AgentCoreError; + expect(fetch).toHaveBeenCalledTimes(1); + expect(error.errorType).toBeUndefined(); + for (const secret of [currentKey, storedKey]) { + expect(error.message).not.toContain(secret); + expect(JSON.stringify(error)).not.toContain(secret); + } + return error; +} + +describe("safe E2B deployment error parameters", () => { + it.each(cases)("preserves only the exact $code field on both write methods", async (input) => { + for (const method of ["POST", "PUT"] as const) { + const error = await reject({ ...input, details: { secret: reflected, allocations: 2, pending: 1, current_generation: 4 } }, method); + expect(error).toMatchObject(input); + expect(error.details).toBeUndefined(); + } + }); + it.each(cases)("preserves $code when PUT omits its stored key", async (input) => { + expect(await reject(input, "PUT", false)).toMatchObject(input); + }); + it.each(cases)("drops mismatched or reflected parameters for $code", async (input) => { + for (const param of [undefined, null, "", "e2b", "e2b.api_key.extra", input.param === "e2b.api_key" ? "e2b.template" : "e2b.api_key", currentKey, storedKey, [input.param], { field: input.param }]) { + for (const includeKey of [true, false]) { + expect(await reject({ ...input, param }, "PUT", includeKey)).toMatchObject({ status: input.status, code: input.code, message: input.message, param: null }); + } + } + }); + it.each(cases)("drops an otherwise valid $code field under the wrong status", async (input) => { + for (const status of [400, 409, 503].filter(value => value !== input.status)) { + expect(await reject({ ...input, status })).toMatchObject({ status, code: input.code, message: input.message, param: null }); + } + }); + it.each(["e2b.api_key", "e2b.template", reflected])("keeps unconfirmed requests unscoped despite parameter %s", async (param) => { + expect(await reject({ status: 503, code: "e2b_request_unconfirmed", param })).toMatchObject({ code: "e2b_request_unconfirmed", param: null }); + }); + it.each(["unknown", currentKey, storedKey, "constructor", "__proto__", "toString"])("replaces unknown credential-bearing code %s without replay", async (code) => { + for (const includeKey of [true, false]) { + const error = await reject({ status: 400, code, param: "e2b.api_key", details: { secret: reflected } }, "PUT", includeKey); + expect(error).toMatchObject({ code: "sandbox_configuration_unconfirmed" }); + expect(error.param).toBeUndefined(); + expect(error.details).toBeUndefined(); + } + }); + it.each([ + { code: "sandbox_generation_stale", details: { current_generation: 4, allocations: 2, pending: 1, secret: reflected }, expected: { current_generation: 4 } }, + { code: "sandbox_in_use", details: { current_generation: 4, allocations: 2, pending: 1, secret: reflected }, expected: { allocations: 2, pending: 1 } }, + { code: "sandbox_in_use", details: { allocations: -1, pending: reflected }, expected: undefined }, + ])("retains the numeric detail allowlist for $code", async ({ code, details, expected }) => { + const error = await reject({ status: 409, code, param: "e2b.api_key", details }); + expect(error.param).toBeNull(); + expect(error.details).toEqual(expected); + }); +}); From 531d1b9e2a3c23cd2ee9f2d99c4982010eaa576f Mon Sep 17 00:00:00 2001 From: SaladDay <1203511142@qq.com> Date: Mon, 28 Sep 2026 10:10:02 +0000 Subject: [PATCH 3/5] Isolate deployment default resolution fixture from SDK database --- .../deployment_model_providers_http_test.go | 21 ++++++++++++++++++- 1 file changed, 20 insertions(+), 1 deletion(-) diff --git a/services/agents-api/internal/store/deployment_model_providers_http_test.go b/services/agents-api/internal/store/deployment_model_providers_http_test.go index 45539b22c..4f8cd9a2d 100644 --- a/services/agents-api/internal/store/deployment_model_providers_http_test.go +++ b/services/agents-api/internal/store/deployment_model_providers_http_test.go @@ -301,7 +301,7 @@ func TestNoneSessionRetryAfterDeploymentDefaultChanges(t *testing.T) { } func TestDeploymentProviderResolutionPairsRevisionDuringReplacement(t *testing.T) { - st, pool := store.NewModelTestStore(t) + st, pool := store.NewManagedTestStore(t) tenant, token := uuid.NewString(), uuid.NewString() auth, err := newTestAuthenticator([]testAPIKey{{OrganizationID: "test-org", ProjectID: uuid.NewString(), SubjectKind: "service_account", SubjectID: "tuple-test", TokenSHA256: device.HashCredential(token), TenantID: tenant}}) if err != nil { @@ -356,3 +356,22 @@ func TestDeploymentProviderResolutionPairsRevisionDuringReplacement(t *testing.T } } } + +// The official-client job starts a fresh server with an independent encryption +// key after the Go suite. Deployment-wide fixtures must leave its database intact. +func TestDeploymentProviderResolutionFixtureIsolation(t *testing.T) { + _, shared := store.NewTestStore(t) + revisions := func() string { + t.Helper() + var value string + if err := shared.QueryRow(t.Context(), "SELECT COALESCE(jsonb_object_agg(harness, revision), '{}'::jsonb)::text FROM deployment_model_providers").Scan(&value); err != nil { + t.Fatal(err) + } + return value + } + before := revisions() + t.Run("resolution", TestDeploymentProviderResolutionPairsRevisionDuringReplacement) + if revisions() != before { + t.Fatal("deployment provider fixture changed the shared test database") + } +} From 759b860cff89c843a6b31121c301a3220144b808 Mon Sep 17 00:00:00 2001 From: SaladDay <1203511142@qq.com> Date: Mon, 28 Sep 2026 10:42:40 +0000 Subject: [PATCH 4/5] test(web): supply nullable provider observations in onboarding fixture --- apps/web/src/features/overview/getting-started.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/web/src/features/overview/getting-started.test.ts b/apps/web/src/features/overview/getting-started.test.ts index 0fcba8d80..fe62021a9 100644 --- a/apps/web/src/features/overview/getting-started.test.ts +++ b/apps/web/src/features/overview/getting-started.test.ts @@ -11,7 +11,7 @@ const deployment = (overrides: Partial = {}): SandboxDeployme }); const fleet = (value: SandboxDeployment, nodes = [node("n1")]): FleetState => ({ status: "ready", snapshot: { deployment: value, nodes, allocations: [], loadedAt: 0 }, refreshing: false, error: null }); const sandboxes = (state: FleetState) => gettingStartedSteps({ sandboxReset: false, fleet: state, localOnly: false, projects: [], sessions: 0, harnesses: [] }).sandboxes; -const provider = { object: "core.model_provider", protocol: "responses", base_url: "https://model.example/v1", api_key_configured: true, updated_at: "2026-09-25T00:00:00Z" } as const; +const provider = { object: "core.model_provider", protocol: "responses", base_url: "https://model.example/v1", api_key_configured: true, last_used_at: null, last_error_code: null, last_error_at: null, updated_at: "2026-09-25T00:00:00Z" } as const; const harness = (id: CoreHarness["id"], fields: Partial = {}): CoreHarness => ({ object: "core.harness", id, enabled: true, default: false, model_provider: null, ...fields }); describe("Getting started steps", () => { From 8ab68e08bd0bb423dae7ab1d6ec379c298d17fe2 Mon Sep 17 00:00:00 2001 From: SaladDay <1203511142@qq.com> Date: Mon, 28 Sep 2026 11:20:54 +0000 Subject: [PATCH 5/5] Complete provider observation fields in browser fixtures --- apps/web/e2e/fixture-console.mjs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/apps/web/e2e/fixture-console.mjs b/apps/web/e2e/fixture-console.mjs index 0c5b51198..7754e01cd 100644 --- a/apps/web/e2e/fixture-console.mjs +++ b/apps/web/e2e/fixture-console.mjs @@ -118,7 +118,7 @@ function reset(mode = "login", fresh = false, sandbox = "configured", nodes = "d // The demo deployment's default harness has a default model; a fresh install has none. harnesses: { claude_sdk: { enabled: true, default: false, provider: null }, - codex: { enabled: true, default: true, provider: fresh ? null : { object: "core.model_provider", harness: "codex", protocol: "responses", base_url: "https://model.example/v1", api_key_configured: true, updated_at: new Date((now - 86400) * 1000).toISOString().replace(/\.\d{3}Z$/, "Z") } }, + codex: { enabled: true, default: true, provider: fresh ? null : { object: "core.model_provider", harness: "codex", protocol: "responses", base_url: "https://model.example/v1", api_key_configured: true, last_used_at: null, last_error_code: null, last_error_at: null, updated_at: new Date((now - 86400) * 1000).toISOString().replace(/\.\d{3}Z$/, "Z") } }, mcode: { enabled: false, default: false, provider: null }, }, // "none": the deployment is not configured yet, so the Nodes page offers setup. @@ -534,6 +534,7 @@ async function harnessRoute(request, response, path) { object: "core.model_provider", harness, protocol: input.protocol, base_url: input.base_url, api_key_configured: true, ...(input.context_window ? { context_window: input.context_window } : {}), ...(input.max_output_tokens ? { max_output_tokens: input.max_output_tokens } : {}), + last_used_at: null, last_error_code: null, last_error_at: null, updated_at: new Date().toISOString().replace(/\.\d{3}Z$/, "Z"), }; return send(response, 200, entry.provider);