fix(core): revoke live stream access and settle failed preparation

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. */}

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

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:

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": "1c789993f75bb380f84f7f82fa366a95625fcdbe830c33e8fde3e2c16b4248fc",
+    "docs/api/public-agent-api.md": "00979732412a971013e8f01b4c74820ff51aafdded6b0f105d25a78af86a6627",
     "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": "33f5700b05c6ffd4ab3a21c5804055893341b7cce8e90bb190bf36d193a887de",
+    "content/docs/sessions.mdx": "bb63d799ed90038836d652f3f866d0309822425114c6b2aa36654b8a991947f6",
     "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",

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

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,15 +10,15 @@
     "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",
     "content/docs/api-reference/files.mdx": "531c835c66b194ab8876c7663b79e42f0090e12420ae32fafeda24e3143740f4",
     "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",

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

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

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:

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

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

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

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