From 786dfce0c85fb39fbc3e7892cc29b72c7d0e0c84 Mon Sep 17 00:00:00 2001 From: saladday <1203511142@qq.com> Date: Wed, 30 Sep 2026 00:07:34 +0800 Subject: [PATCH 1/2] fix(core): revalidate live streams and settle failed preparation --- .../content/docs/api-reference/events.mdx | 4 + .../content/docs/environments-and-files.mdx | 7 ++ apps/docs/content/docs/sessions.mdx | 5 + apps/docs/content/guide-sources.json | 8 +- apps/docs/openapi/public-api.yaml | 4 +- apps/docs/openapi/sources.json | 6 +- contracts/agents-api/environments.md | 7 ++ contracts/agents-api/openapi.yaml | 14 +-- docs/api/public-agent-api.md | 5 + .../internal/api/session_diagnostics.go | 2 + .../internal/api/session_diagnostics_test.go | 2 +- .../internal/api/session_response.go | 2 + services/agents-api/internal/api/stream.go | 24 ++++ .../internal/api/stream_authority_test.go | 104 ++++++++++++++++ .../db/queries/environment_inputs.sql | 4 +- .../db/sqlc/environment_inputs.sql.go | 17 +-- .../internal/execution/preparation.go | 5 + .../internal/execution/worker_schedule.go | 5 +- .../store/environment_input_activity_test.go | 22 ++++ .../internal/store/environment_inputs.go | 12 +- .../store/stream_authority_http_test.go | 112 ++++++++++++++++++ .../store/worker_preparation_failure_test.go | 100 ++++++++++++++++ ..._environment_input_preparation_failure.sql | 11 ++ 23 files changed, 446 insertions(+), 36 deletions(-) create mode 100644 services/agents-api/internal/api/stream_authority_test.go create mode 100644 services/agents-api/internal/store/stream_authority_http_test.go create mode 100644 services/agents-api/internal/store/worker_preparation_failure_test.go create mode 100644 services/agents-api/migrations/000088_environment_input_preparation_failure.sql diff --git a/apps/docs/content/docs/api-reference/events.mdx b/apps/docs/content/docs/api-reference/events.mdx index 227e260ff..f0b0639a7 100644 --- a/apps/docs/content/docs/api-reference/events.mdx +++ b/apps/docs/content/docs/api-reference/events.mdx @@ -26,6 +26,10 @@ _openapi: Session activity includes immutable pending-input connection actions before Turn creation; self_hosted environments use the same safe output as Session retrieval. + + Active streams revalidate the original Project key every second before + output; revocation, Project archival or authentication unavailability + closes the stream. Authentication checks use a five-second timeout. --- {/* 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/content/docs/environments-and-files.mdx b/apps/docs/content/docs/environments-and-files.mdx index 217ef6de3..3d5fb64b2 100644 --- a/apps/docs/content/docs/environments-and-files.mdx +++ b/apps/docs/content/docs/environments-and-files.mdx @@ -239,6 +239,13 @@ is rejected while input is pending and changes nothing. A terminal reservation r affect a later reservation or Turn. Evaluate deadlines after acquiring the Session lock, and return terminal storage outcomes without rolling their transaction back. +A validated Runtime `failed` response with `preparation_failed` and no Run settles +the pending input immediately with `runtime_preparation_failed`. It records a +safe Session failure before any Turn exists and releases the input gate; fixing +the local cause allows new input. Transport loss, capacity rejection and +unconfirmed cleanup remain retryable within the original deadline. Core uses +these common control states, never Harness-specific error text. + Initial messages for a newly created Environment-bearing Session use that same reservation in the creation transaction, including its connection-action event. The creation winner alone inserts it; the stream cursor still precedes that diff --git a/apps/docs/content/docs/sessions.mdx b/apps/docs/content/docs/sessions.mdx index 4ca869b5d..d8bbf2f9f 100644 --- a/apps/docs/content/docs/sessions.mdx +++ b/apps/docs/content/docs/sessions.mdx @@ -324,6 +324,11 @@ cancel while idle does nothing. events sent while you were disconnected are not replayed. Open it before sending input, and recover gaps from [Turns and Items](#turns-and-items). +Open streams recheck the original Project key once per second, including before +output. Revocation or Project archival closes the stream; authentication failure +also closes it. Rechecks use the normal five-second authentication timeout and +send no Session data while waiting. Bytes already sent cannot be recalled. + ```python with client.beta.agents.sessions.events.stream(session.id) as stream: for event in stream: diff --git a/apps/docs/content/guide-sources.json b/apps/docs/content/guide-sources.json index 10e722312..f2d8ef231 100644 --- a/apps/docs/content/guide-sources.json +++ b/apps/docs/content/guide-sources.json @@ -11,9 +11,9 @@ "docs/getting-started/quickstart.md": "b989682ac2e58d59a794118fd371b37d1ea64c957d3512ff53739458a95d0f9a", "docs/api/README.md": "ef2138c892b4de676e7b2b5622b46421204679e656dff7dcc181d261ef22009b", "contracts/agents-api/execution-tools.md": "e0b61f6c0c236c362186c5f6d0ad69a16dd8a8d9afabe1329a1fe22e5f4a71e9", - "docs/api/public-agent-api.md": "5531748f97a84a5e183838cf6fd6ac9bc0c7ae169fb4c6da73a420dbcc77e255", + "docs/api/public-agent-api.md": "37f1da1a3f8364a219cf3075e2b24ce0cea360361dc7f621f816099c750c6604", "docs/examples.md": "0e1bdaeff51c9c36779f817be31ea9816b7d8cb2290cf8350d3c4801f436f4f9", - "contracts/agents-api/environments.md": "9c0e794738c017e9dde57a659dc184d347e33730103583534c43e86d5d4183fd", + "contracts/agents-api/environments.md": "d732a0456e54dc507c94d440cd32055670e341cb8248e40767785129982bd789", "docs/getting-started/nodes.md": "13e1f1e867d3e361ff860960ceacc2dd5806e64f4176cd33f5e49f9a9f4d57e4", "docs/getting-started/self-hosted.md": "4591e05bfa7657f647bedecd9db07b7086acd94b5548511c88f7e5e3567fdb11", "docs/self-hosted-native.md": "1d08e46b8406411f0efeab5940cbda907cd55afb8aec159f5233d2c0ef9d3d54", @@ -41,9 +41,9 @@ "content/docs/quickstart.mdx": "d0ab9537ab68f6dd3109353e937a29c5d58ec3894b56a52873c9a1d79dfa60a6", "content/docs/public-api.mdx": "07a8bd7309f1f5ffa2eaa7edb5823ff9dedb3d2ad728a02112febf1c92526b27", "content/docs/agents-and-tools.mdx": "b8a85fe92e19a71b7ad7c9357ce7ad0595d50c9190edb748ca834df615b056cb", - "content/docs/sessions.mdx": "c5702840e8395545552da35162724019bb0e4ae2d131a57bd71c615a75e4fe00", + "content/docs/sessions.mdx": "d86b4ac428cd8dd0cf8d41370a5ccb1ac5802728979fe63a1b0bdddb768e0629", "content/docs/examples.mdx": "587061e65ab2e841d14980539ba94e2216d4e8135c948a852b9a8bb0359c85d7", - "content/docs/environments-and-files.mdx": "8da8ba45d3c76cc016b1f640d0f3af6e6e8922e4ae0f42ba789443fe59177b9c", + "content/docs/environments-and-files.mdx": "af7b08c06a7df2cb331ff9a8dc9223b510e603348a50f0a85f40959da270339c", "content/docs/hosted-providers.mdx": "06f0b13b053965bbdd1bfb0a7e85d19b50987841e0512fa8331fc9a67ff71fe0", "content/docs/self-hosted-execution.mdx": "161a9000634b4075a55960556322a1b535d8157ba0a4abbeaa84c85c7292ad33", "content/docs/self-hosted-native.mdx": "b2c5b2bb2080b5338bbaa3469f196555be31089ee713d2b9f1be8d47fa7e29cc", diff --git a/apps/docs/openapi/public-api.yaml b/apps/docs/openapi/public-api.yaml index a42d76e65..4e3267064 100644 --- a/apps/docs/openapi/public-api.yaml +++ b/apps/docs/openapi/public-api.yaml @@ -1417,7 +1417,9 @@ paths: - Artifacts /agents/sessions/{session_id}/events: get: - description: Live-only events, including command output fragments from capable Codex peers as agent.output.command_execution_output.delta with stable Item/output indexes. Native text conversion and output quotas apply; completion snapshots remain authoritative. Reconnect through Session, Turn and Items reads; missed events are not replayed. A lagging stream closes with an error when its bounded buffer is exceeded. When a hosted Environment fails to provision, the stream sends agent.session.environment.failed, an error event (environment_error/sandbox_error with the safe step and exit-status reason, never command output) and agent.session.failed, then ends. Session activity includes immutable pending-input connection actions before Turn creation; self_hosted environments use the same safe output as Session retrieval. + description: |- + Live-only events, including command output fragments from capable Codex peers as agent.output.command_execution_output.delta with stable Item/output indexes. Native text conversion and output quotas apply; completion snapshots remain authoritative. Reconnect through Session, Turn and Items reads; missed events are not replayed. A lagging stream closes with an error when its bounded buffer is exceeded. When a hosted Environment fails to provision, the stream sends agent.session.environment.failed, an error event (environment_error/sandbox_error with the safe step and exit-status reason, never command output) and agent.session.failed, then ends. Session activity includes immutable pending-input connection actions before Turn creation; self_hosted environments use the same safe output as Session retrieval. + Active streams revalidate the original Project key every second before output; revocation, Project archival or authentication unavailability closes the stream. Authentication checks use a five-second timeout. parameters: - schema: type: string diff --git a/apps/docs/openapi/sources.json b/apps/docs/openapi/sources.json index 23c76d8f8..d5b1fcfd6 100644 --- a/apps/docs/openapi/sources.json +++ b/apps/docs/openapi/sources.json @@ -1,6 +1,6 @@ { "sources": { - "contracts/agents-api/openapi.yaml": "b706abf259542b7b9a833b566f20048412d3cbc27c4c4284ed6baf5edc605489", + "contracts/agents-api/openapi.yaml": "c68be89861971f10f01e129a3fbed1b18ad177516959acdce15339863ce65de9", "contracts/agents-api/core.openapi.yaml": "5f17f3049a0892888bf4abe919b791b2df45c161b5fa4ee6e27efae2145c7d01", "contracts/agents-api/runtime.openapi.yaml": "505286d5eacf94a14f9527fcf4e7beb4fba4f792681be26ba966254cbf49b4aa" }, @@ -10,7 +10,7 @@ "content/docs/api-reference/environment-templates.mdx": "68df9276e858e525e15bb88af624fc8adee8bcdcc03a788faaff56b560fd038c", "content/docs/api-reference/sessions.mdx": "7c12ac70dfdfef86b5d08b9151a7fafdbec6df4848760c81f059b6a331db81a1", "content/docs/api-reference/artifacts.mdx": "8df8d008126b49b1df88f28a676c4b237ccffc45a470b35ec30eab2daf1c0f7f", - "content/docs/api-reference/events.mdx": "83d7f56a788d073e0b19f0bc9197f39ec364375b9cc3a7ec69b79b54bc71e3eb", + "content/docs/api-reference/events.mdx": "fa509794f3c208c0cdcdaa68615c26b7e400ea6c892922a4e0a6a833e42fe5dd", "content/docs/api-reference/items.mdx": "fa4d248ac5623beb3f00a4ebb5a6c4ff477865ebee351e0c543a970a7ec3654d", "content/docs/api-reference/subagents.mdx": "90e2a1a9129111332e36715de2078ee595590c4c25c7b355962e4185bdeb5388", "content/docs/api-reference/turns.mdx": "46ee9e43aae4ec85b3ee335939dc1e0a238cfda0fb996c6af4d2831bef3377f1", @@ -18,7 +18,7 @@ "content/docs/api-reference/skills.mdx": "774ea05a7a8250fe5af5164388b98dafca99d56dc2eae686e8f628acb4e7c10b", "content/docs/api-reference/vaults.mdx": "d80eade1c2f8100030866abbb0fb6b7a09c94250a859c97cdf1abce32fdfdf1e", "content/docs/api-reference/credentials.mdx": "f129a51f74a3c7021c016600518e2350023f8d29571666ac5d64b7584cd7c7d0", - "openapi/public-api.yaml": "a546a465b7e8e0206a0b60f28533a1ee78f21417c7b54bcdb6e98d7b3638d940", + "openapi/public-api.yaml": "5439ed8aa9109dea4c07718302d6612d600a78f110068ccd3fcd8c6f89eb2b61", "content/docs/api-reference/meta.json": "56bf1dc145ccc8adde38c5f956f1a06f18ef18b646122250d01a294c516f8f83", "content/docs/api-reference/index.mdx": "9fd6b1c54149874d2999235067c101326f9e8a2e707a70b6672c81eb3ebfe63f", "content/docs/api-reference/core/core-administration.mdx": "5fc3ada51d2190a141f13738151e4d513390cc3bd66ce6f5013994bda68569b0", diff --git a/contracts/agents-api/environments.md b/contracts/agents-api/environments.md index e1565139b..b9f7928d7 100644 --- a/contracts/agents-api/environments.md +++ b/contracts/agents-api/environments.md @@ -236,6 +236,13 @@ is rejected while input is pending and changes nothing. A terminal reservation r affect a later reservation or Turn. Evaluate deadlines after acquiring the Session lock, and return terminal storage outcomes without rolling their transaction back. +A validated Runtime `failed` response with `preparation_failed` and no Run settles +the pending input immediately with `runtime_preparation_failed`. It records a +safe Session failure before any Turn exists and releases the input gate; fixing +the local cause allows new input. Transport loss, capacity rejection and +unconfirmed cleanup remain retryable within the original deadline. Core uses +these common control states, never Harness-specific error text. + Initial messages for a newly created Environment-bearing Session use that same reservation in the creation transaction, including its connection-action event. The creation winner alone inserts it; the stream cursor still precedes that diff --git a/contracts/agents-api/openapi.yaml b/contracts/agents-api/openapi.yaml index e5c4c4875..6e0e23607 100644 --- a/contracts/agents-api/openapi.yaml +++ b/contracts/agents-api/openapi.yaml @@ -3553,17 +3553,9 @@ paths: - Artifacts /agents/sessions/{session_id}/events: get: - description: Live-only events, including command output fragments from capable - Codex peers as agent.output.command_execution_output.delta with stable Item/output - indexes. Native text conversion and output quotas apply; completion snapshots - remain authoritative. Reconnect through Session, Turn and Items reads; missed - events are not replayed. A lagging stream closes with an error when its bounded - buffer is exceeded. When a hosted Environment fails to provision, the stream - sends agent.session.environment.failed, an error event (environment_error/sandbox_error - with the safe step and exit-status reason, never command output) and agent.session.failed, - then ends. Session activity includes immutable pending-input connection actions - before Turn creation; self_hosted environments use the same safe output as - Session retrieval. + description: |- + Live-only events, including command output fragments from capable Codex peers as agent.output.command_execution_output.delta with stable Item/output indexes. Native text conversion and output quotas apply; completion snapshots remain authoritative. Reconnect through Session, Turn and Items reads; missed events are not replayed. A lagging stream closes with an error when its bounded buffer is exceeded. When a hosted Environment fails to provision, the stream sends agent.session.environment.failed, an error event (environment_error/sandbox_error with the safe step and exit-status reason, never command output) and agent.session.failed, then ends. Session activity includes immutable pending-input connection actions before Turn creation; self_hosted environments use the same safe output as Session retrieval. + Active streams revalidate the original Project key every second before output; revocation, Project archival or authentication unavailability closes the stream. Authentication checks use a five-second timeout. parameters: - description: agents=v1 in: header diff --git a/docs/api/public-agent-api.md b/docs/api/public-agent-api.md index 7f680aad1..8c4735c75 100644 --- a/docs/api/public-agent-api.md +++ b/docs/api/public-agent-api.md @@ -321,6 +321,11 @@ cancel while idle does nothing. events sent while you were disconnected are not replayed. Open it before sending input, and recover gaps from [Turns and Items](#turns-and-items). +Open streams recheck the original Project key once per second, including before +output. Revocation or Project archival closes the stream; authentication failure +also closes it. Rechecks use the normal five-second authentication timeout and +send no Session data while waiting. Bytes already sent cannot be recalled. + ```python with client.beta.agents.sessions.events.stream(session.id) as stream: for event in stream: diff --git a/services/agents-api/internal/api/session_diagnostics.go b/services/agents-api/internal/api/session_diagnostics.go index ea327a0b3..c3c9812e8 100644 --- a/services/agents-api/internal/api/session_diagnostics.go +++ b/services/agents-api/internal/api/session_diagnostics.go @@ -95,6 +95,8 @@ func (h *Handler) getSessionDiagnostics(w http.ResponseWriter, r *http.Request) failure.Code = "environment_connection_timeout" case "environment_unavailable": failure.Code = "environment_unavailable" + case "runtime_preparation_failed": + failure.Code = "runtime_preparation_failed" case "model_provider_required": failure.Code = "model_provider_required" } diff --git a/services/agents-api/internal/api/session_diagnostics_test.go b/services/agents-api/internal/api/session_diagnostics_test.go index 74fd5e445..c76309dd1 100644 --- a/services/agents-api/internal/api/session_diagnostics_test.go +++ b/services/agents-api/internal/api/session_diagnostics_test.go @@ -79,7 +79,7 @@ func TestDiagnosticsFailurePrecedenceAndUnknownTime(t *testing.T) { for _, tc := range []struct { activity *store.EnvironmentInputActivity code, source string - }{{nil, "harness_error", "turn"}, {&store.EnvironmentInputActivity{Status: "failed"}, "environment_connection_timeout", "environment_input"}, {&store.EnvironmentInputActivity{Status: "failed", Failure: "model_provider_required"}, "model_provider_required", "environment_input"}, {&store.EnvironmentInputActivity{Status: "failed", Failure: "secret-canary"}, "internal_error", "environment_input"}} { + }{{nil, "harness_error", "turn"}, {&store.EnvironmentInputActivity{Status: "failed"}, "environment_connection_timeout", "environment_input"}, {&store.EnvironmentInputActivity{Status: "failed", Failure: "model_provider_required"}, "model_provider_required", "environment_input"}, {&store.EnvironmentInputActivity{Status: "failed", Failure: "runtime_preparation_failed"}, "runtime_preparation_failed", "environment_input"}, {&store.EnvironmentInputActivity{Status: "failed", Failure: "secret-canary"}, "internal_error", "environment_input"}} { value := base value.EnvironmentInputActivity = tc.activity h, _, _ := adminTestHandler(t, func(h *Handler) { h.store = diagnosticSnapshotStore{session: value} }) diff --git a/services/agents-api/internal/api/session_response.go b/services/agents-api/internal/api/session_response.go index cf563d792..c72ae49c7 100644 --- a/services/agents-api/internal/api/session_response.go +++ b/services/agents-api/internal/api/session_response.go @@ -82,6 +82,8 @@ func sessionResponse(session store.Session, executorURL string) (v1.Session, err switch activity.Failure { case "environment_unavailable": message = "The environment is no longer available for this input." + case "runtime_preparation_failed": + message = "Runtime preparation failed before execution. Check the daemon logs and installed capabilities, then submit new input." case "model_provider_required": message = "This Session was created without a model provider and cannot run. Create a new Session with x_agents_core.model_provider or an Agent that has one saved." } diff --git a/services/agents-api/internal/api/stream.go b/services/agents-api/internal/api/stream.go index f4923f311..3814fb72e 100644 --- a/services/agents-api/internal/api/stream.go +++ b/services/agents-api/internal/api/stream.go @@ -10,6 +10,7 @@ import ( "time" v1 "github.com/MiniMax-AI-Dev/parsar/contracts/agents-api/v1" + "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/identity" "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/store" "github.com/go-chi/chi/v5" "github.com/google/uuid" @@ -22,6 +23,7 @@ type eventStore interface { // @Summary Stream live Session events // @Description Live-only events, including command output fragments from capable Codex peers as agent.output.command_execution_output.delta with stable Item/output indexes. Native text conversion and output quotas apply; completion snapshots remain authoritative. Reconnect through Session, Turn and Items reads; missed events are not replayed. A lagging stream closes with an error when its bounded buffer is exceeded. When a hosted Environment fails to provision, the stream sends agent.session.environment.failed, an error event (environment_error/sandbox_error with the safe step and exit-status reason, never command output) and agent.session.failed, then ends. Session activity includes immutable pending-input connection actions before Turn creation; self_hosted environments use the same safe output as Session retrieval. +// @Description Active streams revalidate the original Project key every second before output; revocation, Project archival or authentication unavailability closes the stream. Authentication checks use a five-second timeout. // @Tags Events // @Produce text/event-stream // @Security BearerAuth @@ -76,6 +78,25 @@ func (h *Handler) serveSessionEvents(w http.ResponseWriter, r *http.Request, eve if write == nil { return } + principal := r.Context().Value(principalContextKey{}).(identity.Principal) + var checkedAuthority time.Time + authorized := func() bool { + if time.Since(checkedAuthority) < time.Second { + return true + } + current, ok, err := h.resolvePrincipal(r) + checkedAuthority = time.Now() + return err == nil && ok && current == principal + } + // Recheck on idle polls and before output, including a continuously busy drain. + // The existing resolver bounds authentication calls and fails closed. + streamWrite := write + write = func(data []byte) error { + if !authorized() { + return errors.New("stream authority is no longer available") + } + return streamWrite(data) + } emit := func(event v1.SessionEvent) error { return emitSessionEvent(write, id, event) } if initial != nil { if err := emit(*initial); err != nil { @@ -90,6 +111,9 @@ func (h *Handler) serveSessionEvents(w http.ResponseWriter, r *http.Request, eve recheck, limit := true, int64(-1) var checked time.Time for { + if !authorized() { + return + } changes, err := events.ListSessionEvents(r.Context(), tenant, id, cursor) if errors.Is(err, store.ErrNotFound) { return diff --git a/services/agents-api/internal/api/stream_authority_test.go b/services/agents-api/internal/api/stream_authority_test.go new file mode 100644 index 000000000..ce1418f81 --- /dev/null +++ b/services/agents-api/internal/api/stream_authority_test.go @@ -0,0 +1,104 @@ +package api + +import ( + "context" + "encoding/json" + "errors" + "io" + "net/http" + "net/http/httptest" + "sync" + "sync/atomic" + "testing" + "time" + + v1 "github.com/MiniMax-AI-Dev/parsar/contracts/agents-api/v1" + "github.com/MiniMax-AI-Dev/parsar/internal/agentdaemon/device" + "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/store" + "github.com/google/uuid" +) + +type streamAuthorityResolver struct { + ProjectAPIKeyResolver + unavailable atomic.Bool + calls atomic.Int32 +} + +func (r *streamAuthorityResolver) ResolveProjectAPIKey(ctx context.Context, digest string) (store.ProjectAPIKeyBinding, error) { + r.calls.Add(1) + if r.unavailable.Load() { + return store.ProjectAPIKeyBinding{}, errors.New("resolver unavailable") + } + return r.ProjectAPIKeyResolver.ResolveProjectAPIKey(ctx, digest) +} + +type busyAuthorityStream struct { + *streamFixture + sequence int64 +} + +func (s *busyAuthorityStream) ListSessionEvents(ctx context.Context, _, _ string, _ int64) ([]store.SessionChange, error) { + if ctx.Err() != nil { + return nil, ctx.Err() + } + s.sequence++ + return []store.SessionChange{{Sequence: s.sequence + 10, Event: v1.SessionEvent{Type: "agent.session.idle", EventID: "busy"}}}, nil +} + +func TestBusyStreamRechecksAuthorityAndFailsClosed(t *testing.T) { + key := callerBinding() + key.TokenSHA256 = device.HashCredential("stream") + auth, err := NewAuthenticator([]APIKey{key}) + if err != nil { + t.Fatal(err) + } + resolver := &streamAuthorityResolver{ProjectAPIKeyResolver: auth.keys} + auth.keys = resolver + f := &busyAuthorityStream{streamFixture: &streamFixture{session: store.Session{ID: uuid.NewString(), TenantID: key.TenantID, CreatedAt: time.Now(), Metadata: map[string]string{}, Configuration: json.RawMessage(`{"agent":{"id":"agent_fixture","model":"fixture","tools":[]},"environment":{"type":"none"}}`)}}} + h, err := NewHandler(f, auth, "codex") + if err != nil { + t.Fatal(err) + } + server := httptest.NewServer(h) + defer server.Close() + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + r, _ := http.NewRequestWithContext(ctx, "GET", server.URL+"/v1/agents/sessions/"+f.session.ID+"/events", nil) + r.Header.Set("Authorization", "Bearer stream") + r.Header.Set("OpenAI-Beta", "agents=v1") + response, err := server.Client().Do(r) + if err != nil { + t.Fatal(err) + } + defer response.Body.Close() + if response.StatusCode != 200 { + t.Fatal("stream status", response.StatusCode) + } + done := make(chan error, 1) + var once sync.Once + // Fail the resolver only after receiving real event data. + go func() { + buffer := make([]byte, 4096) + for { + n, err := response.Body.Read(buffer) + if n > 0 { + once.Do(func() { resolver.unavailable.Store(true) }) + } + if err != nil { + done <- err + return + } + } + }() + select { + case err := <-done: + if err != io.EOF { + t.Fatal(err) + } + case <-time.After(3 * time.Second): + t.Fatal("busy stream bypassed authority rechecks") + } + if n := resolver.calls.Load(); n != 3 { + t.Fatal("unexpected per-event authority queries", n) + } +} diff --git a/services/agents-api/internal/db/queries/environment_inputs.sql b/services/agents-api/internal/db/queries/environment_inputs.sql index e39d41236..86f622e5d 100644 --- a/services/agents-api/internal/db/queries/environment_inputs.sql +++ b/services/agents-api/internal/db/queries/environment_inputs.sql @@ -41,7 +41,7 @@ WHERE session_id = $1 AND state = 'pending'; UPDATE environment_input_reservations SET state = 'failed', settled_at = clock_timestamp() WHERE session_id = $1 AND state = 'pending'; --- name: FailEnvironmentInputWithoutModelProvider :execrows +-- name: FailEnvironmentInput :execrows UPDATE environment_input_reservations -SET state = 'failed', settled_at = clock_timestamp(), failure_code = 'model_provider_required' +SET state = 'failed', settled_at = clock_timestamp(), failure_code = sqlc.narg(failure_code) WHERE session_id = @session_id AND id = @id AND state = 'pending'; diff --git a/services/agents-api/internal/db/sqlc/environment_inputs.sql.go b/services/agents-api/internal/db/sqlc/environment_inputs.sql.go index 2a3d7edb2..73621d165 100644 --- a/services/agents-api/internal/db/sqlc/environment_inputs.sql.go +++ b/services/agents-api/internal/db/sqlc/environment_inputs.sql.go @@ -105,19 +105,20 @@ func (q *Queries) ExpireEnvironmentInputReservation(ctx context.Context, arg Exp return err } -const failEnvironmentInputWithoutModelProvider = `-- name: FailEnvironmentInputWithoutModelProvider :execrows +const failEnvironmentInput = `-- name: FailEnvironmentInput :execrows UPDATE environment_input_reservations -SET state = 'failed', settled_at = clock_timestamp(), failure_code = 'model_provider_required' -WHERE session_id = $1 AND id = $2 AND state = 'pending' +SET state = 'failed', settled_at = clock_timestamp(), failure_code = $1 +WHERE session_id = $2 AND id = $3 AND state = 'pending' ` -type FailEnvironmentInputWithoutModelProviderParams struct { - SessionID pgtype.UUID `json:"session_id"` - ID pgtype.UUID `json:"id"` +type FailEnvironmentInputParams struct { + FailureCode pgtype.Text `json:"failure_code"` + SessionID pgtype.UUID `json:"session_id"` + ID pgtype.UUID `json:"id"` } -func (q *Queries) FailEnvironmentInputWithoutModelProvider(ctx context.Context, arg FailEnvironmentInputWithoutModelProviderParams) (int64, error) { - result, err := q.db.Exec(ctx, failEnvironmentInputWithoutModelProvider, arg.SessionID, arg.ID) +func (q *Queries) FailEnvironmentInput(ctx context.Context, arg FailEnvironmentInputParams) (int64, error) { + result, err := q.db.Exec(ctx, failEnvironmentInput, arg.FailureCode, arg.SessionID, arg.ID) if err != nil { return 0, err } diff --git a/services/agents-api/internal/execution/preparation.go b/services/agents-api/internal/execution/preparation.go index f33db9a38..1f82bf31c 100644 --- a/services/agents-api/internal/execution/preparation.go +++ b/services/agents-api/internal/execution/preparation.go @@ -11,6 +11,8 @@ import ( "github.com/google/uuid" ) +var errPreparationFailed = errors.New("runtime preparation failed before admission") + type preparationRejection struct { code string operation string @@ -76,6 +78,9 @@ func (p *preparedStart) observation(env proto.Envelope) (proto.PreparationStatus if status.State == "rejected" { return status, &preparationRejection{code: status.ErrorCode, operation: status.Operation} } + if status.State == "failed" && status.ErrorCode == "preparation_failed" && status.RunID == "" { + return status, errPreparationFailed + } switch status.State { case "preparing", "ready", "starting", "started": return status, nil diff --git a/services/agents-api/internal/execution/worker_schedule.go b/services/agents-api/internal/execution/worker_schedule.go index fbb932320..9528de080 100644 --- a/services/agents-api/internal/execution/worker_schedule.go +++ b/services/agents-api/internal/execution/worker_schedule.go @@ -94,7 +94,10 @@ func (w *Worker) runEnvironmentInput(ctx context.Context, item scheduledWork) er return nil } if errors.Is(err, store.ErrModelProviderRequired) { - return w.dispatcher.Store.FailEnvironmentInputWithoutModelProvider(ctx, item.TenantID, item.SessionID, item.reservationID) + return w.dispatcher.Store.FailEnvironmentInput(ctx, item.TenantID, item.SessionID, item.reservationID, "model_provider_required") + } + if errors.Is(err, errPreparationFailed) && run.Reservation.State == store.EnvironmentInputPending { + return w.dispatcher.Store.FailEnvironmentInput(ctx, item.TenantID, item.SessionID, item.reservationID, "runtime_preparation_failed") } if run.Reservation.State == store.EnvironmentInputAdmitted { return err diff --git a/services/agents-api/internal/store/environment_input_activity_test.go b/services/agents-api/internal/store/environment_input_activity_test.go index df9e42063..54280f2fa 100644 --- a/services/agents-api/internal/store/environment_input_activity_test.go +++ b/services/agents-api/internal/store/environment_input_activity_test.go @@ -251,3 +251,25 @@ func TestEnvironmentInputActivityRecoversWaitingActionAndHidesDeletion(t *testin t.Fatal("deleted activity events remained visible", err) } } + +func TestPreparationFailurePreservesCancelledAndNewerInput(t *testing.T) { + s, _ := testStore(t) + tenant, session := environmentInputSession(t, s) + first := reserveEnvironmentInput(t, s, tenant, session.ID, "cancelled") + if _, err := s.CancelEnvironmentInput(t.Context(), tenant, session.ID, first.ID); err != nil { + t.Fatal(err) + } + next := reserveEnvironmentInput(t, s, tenant, session.ID, "new") + if err := s.FailEnvironmentInput(t.Context(), tenant, session.ID, first.ID, "runtime_preparation_failed"); err != nil { + t.Fatal(err) + } + for id, state := range map[string]string{first.ID: EnvironmentInputCancelled, next.ID: EnvironmentInputPending} { + current, err := s.GetEnvironmentInputReservation(t.Context(), tenant, session.ID, id) + if err != nil || current.State != state { + t.Fatal("late failure changed another outcome", err) + } + } + if err := s.FailEnvironmentInput(t.Context(), tenant, session.ID, next.ID, "secret-canary"); !errors.Is(err, ErrInvalidInput) { + t.Fatal("unclassified diagnostic accepted", err) + } +} diff --git a/services/agents-api/internal/store/environment_inputs.go b/services/agents-api/internal/store/environment_inputs.go index b53f174d4..d2ddd489d 100644 --- a/services/agents-api/internal/store/environment_inputs.go +++ b/services/agents-api/internal/store/environment_inputs.go @@ -174,16 +174,18 @@ func (s *Store) CancelEnvironmentInput(ctx context.Context, tenantID, sessionID, return s.settleEnvironmentInput(ctx, tenantID, sessionID, reservationID, EnvironmentInputCancelled) } -// FailEnvironmentInputWithoutModelProvider settles pending input of a Session -// that has no frozen model provider, reserved before providers were required, -// instead of letting it wait for its deadline. Settled input is unchanged. -func (s *Store) FailEnvironmentInputWithoutModelProvider(ctx context.Context, tenantID, sessionID, reservationID string) error { +// FailEnvironmentInput settles a confirmed pre-admission failure. The Session +// lock and pending-state predicate preserve cancellation and newer input. +func (s *Store) FailEnvironmentInput(ctx context.Context, tenantID, sessionID, reservationID, code string) error { + if code != "model_provider_required" && code != "runtime_preparation_failed" { + return ErrInvalidInput + } id, err := parseID(reservationID) if err != nil { return err } return s.withEnvironmentInputSession(ctx, tenantID, sessionID, func(ctx context.Context, q *sqlc.Queries, session pgtype.UUID) error { - _, err := q.FailEnvironmentInputWithoutModelProvider(ctx, sqlc.FailEnvironmentInputWithoutModelProviderParams{SessionID: session, ID: id}) + _, err := q.FailEnvironmentInput(ctx, sqlc.FailEnvironmentInputParams{SessionID: session, ID: id, FailureCode: pgtype.Text{String: code, Valid: true}}) return err }) } diff --git a/services/agents-api/internal/store/stream_authority_http_test.go b/services/agents-api/internal/store/stream_authority_http_test.go new file mode 100644 index 000000000..9dfc0eb9e --- /dev/null +++ b/services/agents-api/internal/store/stream_authority_http_test.go @@ -0,0 +1,112 @@ +package store_test + +import ( + "bufio" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/adminaudit" + "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/api" + "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/store" + "github.com/google/uuid" +) + +func TestLiveStreamClosesAfterKeyRevocationOrProjectArchive(t *testing.T) { + for _, archive := range []bool{false, true} { + name := "key-revocation" + if archive { + name = "project-archive" + } + t.Run(name, func(t *testing.T) { + s, _ := store.NewTestStore(t) + projectID := uuid.NewString() + ctx := adminaudit.WithSource(t.Context(), adminaudit.Source{CredentialID: "12345678", ActorLabel: "test", RequestID: uuid.NewString(), TraceID: uuid.NewString(), ProjectID: projectID}) + project, err := s.CreateProject(ctx, projectID, "Stream authority") + if err != nil { + t.Fatal(err) + } + reader, err := s.CreateProjectAPIKey(ctx, project.ID, uuid.NewString(), "reader") + if err != nil { + t.Fatal(err) + } + peer, err := s.CreateProjectAPIKey(ctx, project.ID, uuid.NewString(), "peer") + if err != nil { + t.Fatal(err) + } + session, err := s.CreateSession(t.Context(), project.TenantID, store.CreateSessionInput{Creator: store.FixtureCreator(), Engine: "codex", IdempotencyKey: uuid.NewString(), Configuration: json.RawMessage(`{"agent":{"id":"agent_fixture","model":"fixture","tools":[]},"environment":{"type":"self_hosted","workspace_directory":"/workspace"}}`)}) + if err != nil { + t.Fatal(err) + } + auth, err := api.NewDatabaseAuthenticator(s) + if err != nil { + t.Fatal(err) + } + h, err := api.NewHandler(s, auth, "codex", api.WithEnvironmentRemoteURL("wss://core.example/api/v1/agent-daemon/ws")) + if err != nil { + t.Fatal(err) + } + server := httptest.NewServer(h) + defer server.Close() + request := func(key, suffix string) *http.Response { + t.Helper() + r, _ := http.NewRequestWithContext(t.Context(), "GET", server.URL+"/v1/agents/sessions/"+session.ID+suffix, nil) + r.Header.Set("Authorization", "Bearer "+key) + r.Header.Set("OpenAI-Beta", "agents=v1") + response, err := server.Client().Do(r) + if err != nil { + t.Fatal(err) + } + return response + } + response := request(reader.Key, "/events") + defer response.Body.Close() + if response.StatusCode != 200 { + t.Fatal("stream status", response.StatusCode) + } + input := bufio.NewReader(response.Body) + if line, err := input.ReadString('\n'); err != nil || line != ": connected\n" { + t.Fatal("stream not open", err) + } + // Keep an idle stream open across a successful authority recheck. + time.Sleep(1100 * time.Millisecond) + if archive { + _, err = s.ArchiveProject(ctx, project.ID) + } else { + err = s.RevokeProjectAPIKey(ctx, project.ID, reader.ID) + } + if err != nil { + t.Fatal(err) + } + fresh := request(reader.Key, "") + fresh.Body.Close() + if fresh.StatusCode != 401 { + t.Fatal("fresh request retained authority", fresh.StatusCode) + } + done := make(chan error, 1) + go func() { _, err := io.Copy(io.Discard, input); done <- err }() + select { + case err := <-done: + if err != nil { + t.Fatal(err) + } + case <-time.After(3 * time.Second): + t.Fatal("revoked idle stream remained open") + } + if !archive { + valid := request(peer.Key, "") + valid.Body.Close() + if valid.StatusCode != 200 { + t.Fatal("revocation affected peer", valid.StatusCode) + } + // New events remain available to valid callers after the reader has closed. + if _, err := s.ReserveEnvironmentInput(t.Context(), project.TenantID, session.ID, "after-revocation", []store.Input{{Kind: "message", Payload: json.RawMessage(`{"text":"new event"}`)}}); err != nil { + t.Fatal(err) + } + } + }) + } +} diff --git a/services/agents-api/internal/store/worker_preparation_failure_test.go b/services/agents-api/internal/store/worker_preparation_failure_test.go new file mode 100644 index 000000000..7040cb630 --- /dev/null +++ b/services/agents-api/internal/store/worker_preparation_failure_test.go @@ -0,0 +1,100 @@ +package store_test + +import ( + "encoding/json" + "testing" + "time" + + "github.com/MiniMax-AI-Dev/parsar/internal/agentdaemon/proto" + "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/store" +) + +func TestWorkerSettlesConfirmedPreparationFailureAndAcceptsNewInput(t *testing.T) { + h := newDispatchHarnessForSession(t, []byte(`{"agent":{"model":"test-model"},"environment":{"type":"self_hosted","workspace_directory":"/workspace"}}`), false) + enableWorkerEnvironment(t, h) + frames := workerFrames(t, h) + pending, err := h.s.ReserveEnvironmentInput(t.Context(), h.tenant, h.session.ID, "first", []store.Input{{Kind: "message", Payload: json.RawMessage(`{"text":"first"}`)}}) + if err != nil { + t.Fatal(err) + } + _, stop := startEnvironmentExpiryWorker(t, h.d) + defer stop() + prepare := nextWorkerFrame(t, frames, proto.TypeExecutionPrepare) + handle := acknowledgePreparation(h, prepare.ID) + h.write(prepare.ID, proto.TypePreparationStatus, proto.PreparationStatusPayload{Handle: handle, Revision: 2, State: "failed", ErrorCode: "preparation_failed"}) + nextWorkerFrame(t, frames, proto.TypeExecutionRelease) + awaitDaemonRemoteCondition(t, t.Context(), 3*time.Second, "failed reservation settlement", func() bool { + current, err := h.s.GetEnvironmentInputReservation(t.Context(), h.tenant, h.session.ID, pending.ID) + return err == nil && current.State == store.EnvironmentInputFailed + }) + session, err := h.s.GetSession(t.Context(), h.tenant, h.session.ID) + if err != nil || session.PendingInput || session.LastTurn != nil || session.EnvironmentInputActivity == nil || session.EnvironmentInputActivity.Failure != "runtime_preparation_failed" { + t.Fatal("preparation did not release input with a safe failure", err) + } + changes, err := h.s.ListSessionEvents(t.Context(), h.tenant, h.session.ID, 0) + if err != nil { + t.Fatal(err) + } + failures := 0 + for _, change := range changes { + if change.Event.Type == "agent.session.failed" { + failures++ + if !change.Settled || change.EnvironmentInputActivity.Failure != "runtime_preparation_failed" { + t.Fatal("failure event lost settlement") + } + } + } + if failures != 1 { + t.Fatal("failure events", failures) + } + select { + case frame := <-frames: + t.Fatal("failed input retried", frame.Type) + case <-time.After(1200 * time.Millisecond): + } + next, err := h.s.ReserveEnvironmentInput(t.Context(), h.tenant, h.session.ID, "next", []store.Input{{Kind: "message", Payload: json.RawMessage(`{"text":"next"}`)}}) + if err != nil { + t.Fatal("new input remained blocked", err) + } + prepare = nextWorkerFrame(t, frames, proto.TypeExecutionPrepare) + handle = acknowledgePreparation(h, prepare.ID) + h.write(prepare.ID, proto.TypePreparationStatus, proto.PreparationStatusPayload{Handle: handle, Revision: 2, State: "ready"}) + startFrame := nextWorkerFrame(t, frames, proto.TypeExecutionStart) + var start proto.ExecutionStartPayload + if startFrame.DecodePayload(&start) != nil || start.RunID == "" || inputTextForTest(t, start.Input) != "next" { + t.Fatal("new input was not admitted") + } + current, err := h.s.GetEnvironmentInputReservation(t.Context(), h.tenant, h.session.ID, next.ID) + if err != nil || current.State != store.EnvironmentInputAdmitted { + t.Fatal("new input state", err) + } + h.write(start.RunID, proto.TypeDone, proto.DonePayload{Content: "complete"}) +} + +func TestWorkerRetriesUncertainPreparationFailure(t *testing.T) { + for _, code := range []string{"connection_closed", "executor_cleanup_unconfirmed"} { + t.Run(code, func(t *testing.T) { + h := newDispatchHarnessForSession(t, []byte(`{"agent":{"model":"test-model"},"environment":{"type":"self_hosted","workspace_directory":"/workspace"}}`), false) + enableWorkerEnvironment(t, h) + frames := workerFrames(t, h) + pending, err := h.s.ReserveEnvironmentInput(t.Context(), h.tenant, h.session.ID, "retry", []store.Input{{Kind: "message", Payload: json.RawMessage(`{"text":"retry"}`)}}) + if err != nil { + t.Fatal(err) + } + _, stop := startEnvironmentExpiryWorker(t, h.d) + defer stop() + prepare := nextWorkerFrame(t, frames, proto.TypeExecutionPrepare) + handle := acknowledgePreparation(h, prepare.ID) + h.write(prepare.ID, proto.TypePreparationStatus, proto.PreparationStatusPayload{Handle: handle, Revision: 2, State: "failed", ErrorCode: code}) + nextWorkerFrame(t, frames, proto.TypeExecutionRelease) + nextWorkerFrame(t, frames, proto.TypeExecutionPrepare) + current, err := h.s.GetEnvironmentInputReservation(t.Context(), h.tenant, h.session.ID, pending.ID) + if err != nil || current.State != store.EnvironmentInputPending || !current.Deadline.Equal(pending.Deadline) { + t.Fatal("transient failure settled or extended input", err) + } + if _, err := h.s.CancelEnvironmentInput(t.Context(), h.tenant, h.session.ID, pending.ID); err != nil { + t.Fatal(err) + } + }) + } +} diff --git a/services/agents-api/migrations/000088_environment_input_preparation_failure.sql b/services/agents-api/migrations/000088_environment_input_preparation_failure.sql new file mode 100644 index 000000000..06b57a41d --- /dev/null +++ b/services/agents-api/migrations/000088_environment_input_preparation_failure.sql @@ -0,0 +1,11 @@ +-- +goose Up +ALTER TABLE environment_input_reservations + DROP CONSTRAINT environment_input_reservations_check1, + ADD CONSTRAINT environment_input_reservations_check1 + CHECK (failure_code IS NULL OR (state = 'failed' AND failure_code IN ('model_provider_required', 'runtime_preparation_failed'))); + +-- +goose Down +ALTER TABLE environment_input_reservations + DROP CONSTRAINT environment_input_reservations_check1, + ADD CONSTRAINT environment_input_reservations_check1 + CHECK (failure_code IS NULL OR (state = 'failed' AND failure_code = 'model_provider_required')); From 0d30a539004934933fcd2089a786311c3034fb5d Mon Sep 17 00:00:00 2001 From: saladday <1203511142@qq.com> Date: Wed, 30 Sep 2026 00:14:12 +0800 Subject: [PATCH 2/2] test(core): exercise authority loss after live event delivery --- .../internal/api/stream_authority_test.go | 20 +++++++------------ 1 file changed, 7 insertions(+), 13 deletions(-) diff --git a/services/agents-api/internal/api/stream_authority_test.go b/services/agents-api/internal/api/stream_authority_test.go index ce1418f81..a21c19a96 100644 --- a/services/agents-api/internal/api/stream_authority_test.go +++ b/services/agents-api/internal/api/stream_authority_test.go @@ -1,13 +1,12 @@ package api import ( + "bufio" "context" "encoding/json" "errors" - "io" "net/http" "net/http/httptest" - "sync" "sync/atomic" "testing" "time" @@ -75,24 +74,19 @@ func TestBusyStreamRechecksAuthorityAndFailsClosed(t *testing.T) { t.Fatal("stream status", response.StatusCode) } done := make(chan error, 1) - var once sync.Once // Fail the resolver only after receiving real event data. go func() { - buffer := make([]byte, 4096) - for { - n, err := response.Body.Read(buffer) - if n > 0 { - once.Do(func() { resolver.unavailable.Store(true) }) - } - if err != nil { - done <- err - return + scanner := bufio.NewScanner(response.Body) + for scanner.Scan() { + if scanner.Text() == "event: agent.session.idle" { + resolver.unavailable.Store(true) } } + done <- scanner.Err() }() select { case err := <-done: - if err != io.EOF { + if err != nil { t.Fatal(err) } case <-time.After(3 * time.Second):