Fence worker credentials across server restarts

docs/internal/server-secrets-strategy.md

@@ -126,6 +126,7 @@ There is no startup-time secret generation or import of optional integration sec
 - Worker and render-graph subprocesses start from `env_clear()` and re-add only explicit allowlisted variables.
 - Authority-bearing values are re-injected intentionally. For worker subprocesses this is `FABRO_WORKER_TOKEN`, plus any explicitly required internal value such as a vault-derived `GITHUB_APP_PRIVATE_KEY`; it is not user auth state such as `FABRO_DEV_TOKEN` or `auth.json`.
 - The worker reads `FABRO_WORKER_TOKEN` from its env at startup (in `main()` before Tokio initializes) and immediately calls `std::env::remove_var` to scrub it. The token then flows through function arguments to `runner::execute`. Every descendant process (hooks, sandbox commands, MCP stdio, etc.) therefore inherits a worker env that no longer contains the bearer, so an unscrubbed spawn site cannot leak it.
+- Worker tokens name the issuing server instance in their JWT audience. A new server instance rejects tokens from the previous instance before handling any request, even when `SESSION_SECRET` is unchanged. Startup recovery replaces in-flight workers; an old worker that survives the restart must not publish platform lifecycle records or reacquire a released Petri lease. This audience is a random, non-secret instance identifier, not a new bootstrap secret. User credentials keep their existing lifetime.
 - The daemon child inherits the parent env unchanged except for output-format hygiene (`FABRO_JSON` removal).
 
 ## Tests

lib/apps/fabro-cli/tests/it/support/auth_tokens.rs

@@ -63,6 +63,7 @@ struct TestJwtClaims {
 #[derive(serde::Serialize)]
 struct WorkerTokenClaims {
     iss:    String,
+    aud:    String,
     iat:    u64,
     exp:    u64,
     run_id: String,
@@ -138,6 +139,10 @@ fn derived_jwt_key() -> [u8; 32] {
     key
 }
 
+#[expect(
+    clippy::disallowed_methods,
+    reason = "Synchronous CLI test setup reads the isolated server audience before spawning a worker."
+)]
 pub(crate) fn issue_test_worker_jwt(storage_dir: &Path, run_id: &str) -> String {
     let runtime_directory = Storage::new(storage_dir).runtime_directory();
     let session_secret = envfile::read_env_file(&runtime_directory.env_path())
@@ -155,6 +160,12 @@ pub(crate) fn issue_test_worker_jwt(storage_dir: &Path, run_id: &str) -> String
         .expect("current timestamp should be positive");
     let claims = WorkerTokenClaims {
         iss:    WORKER_TOKEN_ISSUER.to_string(),
+        aud:    std::fs::read_to_string(
+            runtime_directory
+                .record_path()
+                .with_file_name("test-worker-audience"),
+        )
+        .expect("test server should publish its worker token audience"),
         iat:    now,
         exp:    now + WORKER_TOKEN_TTL_SECS,
         run_id: run_id.to_string(),

lib/apps/fabro-server/src/petri_runs.rs

@@ -74,7 +74,7 @@ impl PetriRuns {
 
     /// The writer handle `owner` holds on the run: the one kept from its
     /// open, or a reopen when the store's lease row still names the owner
-    /// (the server restarted, or the open's reply was lost). An owner the
+    /// (the open's reply was lost). An owner the
     /// lease no longer names gets `StaleOwner`.
     pub(crate) async fn writer(
         &self,
@@ -515,6 +515,201 @@ mod tests {
         drop(before);
     }
 
+    /// A crash leaves the old worker alive, but only the replacement may
+    /// publish to the restarted server. Exercise the HTTP boundary before
+    /// the replacement starts, while it runs, and after it has settled.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn restarted_server_fences_stale_completion_and_keeps_resumed_result() {
+        for terminal in [
+            RunStatus::Succeeded {
+                reason: SuccessReason::Completed,
+            },
+            RunStatus::Failed {
+                reason: FailureReason::WorkflowError,
+            },
+        ] {
+            let (store, artifacts) = test_store_bundle();
+            let vault_path = test_secret_store_path();
+            let before = TestAppStateBuilder::new()
+                .store_bundle(Arc::clone(&store), artifacts.clone())
+                .vault_path(vault_path.clone())
+                .vault_entries([(EnvVars::OPENAI_API_KEY, "test-openai-api-key")])
+                .build();
+            let app = build_test_router(Arc::clone(&before));
+            let run_id = create_and_start_petri_run(&app).await;
+            let old_token = before.test_issue_worker_token(&run_id);
+            run_to_running_as_worker(&app, run_id, &old_token).await;
+            let old_handle = before
+                .petri_runs
+                .open(run_id, Access::Create {
+                    owner: OwnerId::new("old-worker"),
+                })
+                .await
+                .expect("the old worker holds the lease");
+
+            let runtime = Arc::new(HeldWorkerRuntime::default());
+            let after = TestAppStateBuilder::new()
+                .store_bundle(store, artifacts)
+                .vault_path(vault_path)
+                .vault_entries([(EnvVars::OPENAI_API_KEY, "test-openai-api-key")])
+                .worker_runtime(Arc::clone(&runtime) as Arc<dyn WorkerRuntime>)
+                .build();
+            let app = build_test_router(Arc::clone(&after));
+            assert_eq!(
+                reconcile_incomplete_runs_on_startup(&after).await.unwrap(),
+                1
+            );
+            assert_stale_completion_rejected(&after, &app, run_id, &old_token).await;
+
+            write_test_server_record(&after);
+            spawn_scheduler(Arc::clone(&after));
+            runtime.wait_for_start().await;
+            assert_eq!(runtime.launched_mode(), Some("resume"));
+            let token = after.test_issue_worker_token(&run_id);
+            let resumed = after
+                .petri_runs
+                .open(run_id, Access::Write {
+                    owner: OwnerId::new("worker-1"),
+                })
+                .await
+                .expect("the replacement takes the lease");
+            run_to_running_as_worker(&app, run_id, &token).await;
+            assert_stale_completion_rejected(&after, &app, run_id, &old_token).await;
+
+            let transition = if matches!(terminal, RunStatus::Succeeded { .. }) {
+                record_finish_as_worker(&app, run_id, &token).await;
+                RunLifecycleKind::Succeeded
+            } else {
+                // A current worker's failure before an engine finish must
+                // still be able to end the run.
+                RunLifecycleKind::Failed
+            };
+            record_lifecycle_as_worker(&app, run_id, &token, transition, terminal).await;
+            assert_stale_completion_rejected(&after, &app, run_id, &old_token).await;
+
+            let response = app
+                .clone()
+                .oneshot(
+                    Request::builder()
+                        .uri(format!("/api/v1/runs/{run_id}/state"))
+                        .body(Body::empty())
+                        .unwrap(),
+                )
+                .await
+                .unwrap();
+            let view: serde_json::Value = fabro_test::expect_axum_json(
+                response,
+                StatusCode::OK,
+                "GET /api/v1/runs/{id}/state after resumed completion",
+            )
+            .await;
+            assert_eq!(view["status"], serde_json::to_value(terminal).unwrap());
+            let response = app
+                .clone()
+                .oneshot(
+                    Request::builder()
+                        .uri(format!("/api/v1/runs/{run_id}"))
+                        .body(Body::empty())
+                        .unwrap(),
+                )
+                .await
+                .unwrap();
+            let summary = fabro_test::expect_axum_json(
+                response,
+                StatusCode::OK,
+                "GET /api/v1/runs/{id} after resumed completion",
+            )
+            .await;
+            assert_eq!(summary["lifecycle"]["status"], view["status"]);
+            assert_eq!(after.test_managed_run_status(&run_id), Some(terminal));
+
+            runtime.end_worker();
+            drop(resumed);
+            drop(old_handle);
+            drop(before);
+        }
+    }
+
+    async fn assert_stale_completion_rejected(
+        state: &AppState,
+        app: &axum::Router,
+        run_id: RunId,
+        token: &str,
+    ) {
+        let view = run_records::projection(state, run_id)
+            .await
+            .unwrap()
+            .unwrap();
+        let managed = state.test_managed_run_status(&run_id);
+        let records = state
+            .stores
+            .run_summaries
+            .platform_records()
+            .read(&run_id)
+            .await
+            .unwrap();
+        let response = app
+            .clone()
+            .oneshot(
+                Request::builder()
+                    .method("POST")
+                    .uri(format!("/api/v1/runs/{run_id}/petri/open"))
+                    .header(header::AUTHORIZATION, format!("Bearer {token}"))
+                    .header(header::CONTENT_TYPE, "application/json")
+                    .body(Body::from(
+                        json!({ "access": "write", "owner": "old-worker" }).to_string(),
+                    ))
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+        fabro_test::assert_axum_status(
+            response,
+            StatusCode::UNAUTHORIZED,
+            "superseded worker cannot reacquire the lease",
+        )
+        .await;
+        for (transition, status) in [
+            (RunLifecycleKind::Failed, RunStatus::Failed {
+                reason: FailureReason::WorkflowError,
+            }),
+            (RunLifecycleKind::Succeeded, RunStatus::Succeeded {
+                reason: SuccessReason::Completed,
+            }),
+        ] {
+            let response = append_lifecycle_as_worker(app, run_id, token, transition, status).await;
+            fabro_test::assert_axum_status(
+                response,
+                StatusCode::UNAUTHORIZED,
+                "superseded worker terminal write",
+            )
+            .await;
+        }
+        let after = run_records::projection(state, run_id)
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(
+            after.status, view.status,
+            "stale completions leave the public result unchanged"
+        );
+        assert_eq!(state.test_managed_run_status(&run_id), managed);
+        assert_eq!(
+            serde_json::to_value(
+                state
+                    .stores
+                    .run_summaries
+                    .platform_records()
+                    .read(&run_id)
+                    .await
+                    .unwrap()
+            )
+            .unwrap(),
+            serde_json::to_value(records).unwrap(),
+            "rejected records never enter the durable log"
+        );
+    }
+
     /// A server whose one worker is held open by the test, with a run the
     /// worker has taken over the API: the run's id and the worker's token.
     async fn held_worker_run(
@@ -558,13 +753,23 @@ mod tests {
         transition: RunLifecycleKind,
         status: RunStatus,
     ) {
+        let response = append_lifecycle_as_worker(app, run_id, token, transition, status).await;
+        fabro_test::assert_axum_status(response, StatusCode::OK, "worker lifecycle append").await;
+    }
+
+    async fn append_lifecycle_as_worker(
+        app: &axum::Router,
+        run_id: RunId,
+        token: &str,
+        transition: RunLifecycleKind,
+        status: RunStatus,
+    ) -> axum::response::Response {
         let record =
             PlatformRecord::RunLifecycle(RunLifecycleRecord::new(transition).with_status(status));
         let body = json!({
             "record": serde_json::to_value(&record).expect("the record encodes"),
         });
-        let response = app
-            .clone()
+        app.clone()
             .oneshot(
                 Request::builder()
                     .method("POST")
@@ -575,8 +780,7 @@ mod tests {
                     .expect("the append request builds"),
             )
             .await
-            .expect("the append request completes");
-        assert_eq!(response.status(), StatusCode::OK);
+            .expect("the append request completes")
     }
 
     /// Petri's own finish of the run, stored the way its worker stores it:

lib/apps/fabro-server/src/principal_middleware.rs

@@ -568,13 +568,12 @@ mod tests {
     use fabro_types::settings::ServerAuthMethod;
     use fabro_types::{AuthMethod, IdpIdentity, RunId};
     use jsonwebtoken::EncodingKey;
-    use uuid::Uuid;
 
     use super::*;
     use crate::auth::{self, AuthErrorCode};
     use crate::worker_token::{
-        WORKER_RUN_TOOLS_SCOPE, WORKER_TOKEN_ISSUER, WORKER_TOKEN_SCOPE, WorkerScopeSet,
-        WorkerTokenClaims, issue_worker_token, worker_token_header,
+        WORKER_RUN_TOOLS_SCOPE, WORKER_TOKEN_SCOPE, WorkerScopeSet, WorkerTokenClaims,
+        issue_worker_token, worker_token_header,
     };
 
     const TEST_JWT_ISSUER: &str = "https://fabro.example";
@@ -635,24 +634,29 @@ mod tests {
         let secret = state
             .server_secret(EnvVars::SESSION_SECRET)
             .expect("test state should have session secret");
-        issue_worker_claims_with_secret(secret.as_bytes(), run_id, exp, scope)
+        issue_worker_claims_with_secret(state, secret.as_bytes(), run_id, exp, scope)
     }
 
     fn issue_worker_claims_with_secret(
+        state: &AppState,
         secret: &[u8],
         run_id: RunId,
         exp: u64,
         scope: &str,
     ) -> String {
         let worker_key = auth::derive_worker_jwt_key(secret).unwrap();
-        let claims = WorkerTokenClaims {
-            iss: WORKER_TOKEN_ISSUER.to_string(),
-            iat: 1,
-            exp,
-            run_id: run_id.to_string(),
-            scope: scope.to_string(),
-            jti: Uuid::new_v4().simple().to_string(),
-        };
+        let keys = state.worker_token_keys();
+        let token = issue_worker_token(keys, &run_id).unwrap();
+        let mut claims = jsonwebtoken::decode::<WorkerTokenClaims>(
+            &token,
+            keys.decoding_key(),
+            keys.validation(),
+        )
+        .unwrap()
+        .claims;
+        claims.iat = 1;
+        claims.exp = exp;
+        claims.scope = scope.to_string();
         jsonwebtoken::encode(
             &worker_token_header(),
             &claims,
@@ -758,6 +762,7 @@ mod tests {
     fn classifies_invalid_worker_jwt_signature_as_invalid() {
         let state = crate::test_support::test_app_state();
         let token = issue_worker_claims_with_secret(
+            state.as_ref(),
             b"other-principal-middleware-secret-0001",
             RunId::new(),
             u64::MAX / 2,

lib/apps/fabro-server/src/serve.rs

@@ -41,6 +41,8 @@ use crate::server::{
 use crate::server_secrets::{ServerSecrets, process_env_snapshot};
 use crate::startup::{resolve_startup, validate_startup_configuration};
 use crate::static_files;
+#[cfg(any(test, feature = "test-support"))]
+use crate::test_support;
 
 pub const DEFAULT_TCP_PORT: u16 = 32276;
 type EnvLookup = Arc<dyn Fn(&str) -> Option<String> + Send + Sync>;
@@ -749,6 +751,8 @@ where
         #[cfg(any(test, feature = "test-support"))]
         automation_materializer_override: None,
     })?;
+    #[cfg(any(test, feature = "test-support"))]
+    test_support::write_test_worker_audience(&state).await?;
     let reconciled = reconcile_incomplete_runs_on_startup(&state).await?;
     if reconciled > 0 {
         info!(

lib/apps/fabro-server/src/server/tests.rs

@@ -523,10 +523,10 @@ async fn http_log_records_user_principal_fields() {
 
 #[tokio::test(flavor = "current_thread")]
 async fn http_log_records_worker_principal_fields() {
-    let (_state, app) = jwt_auth_app();
+    let (state, app) = jwt_auth_app();
     let user_bearer = issue_test_user_jwt();
     let run_id = create_run_with_bearer(&app, &user_bearer).await;
-    let worker_bearer = issue_test_worker_token(&run_id);
+    let worker_bearer = issue_test_worker_token(&state, &run_id);
     let (_guard, events) = capture_server_logs();
 
     let response = app
@@ -663,17 +663,14 @@ fn issue_test_user_jwt() -> String {
     )
 }
 
-fn issue_test_worker_token(run_id: &RunId) -> String {
-    let keys = WorkerTokenKeys::from_master_secret(TEST_SESSION_SECRET.as_bytes())
-        .expect("worker keys should derive");
-    crate::worker_token::issue_worker_token(&keys, run_id).expect("worker token should issue")
+fn issue_test_worker_token(state: &AppState, run_id: &RunId) -> String {
+    crate::worker_token::issue_worker_token(state.worker_token_keys(), run_id)
+        .expect("worker token should issue")
 }
 
-fn issue_test_run_tools_worker_token(run_id: &RunId) -> String {
-    let keys = WorkerTokenKeys::from_master_secret(TEST_SESSION_SECRET.as_bytes())
-        .expect("worker keys should derive");
+fn issue_test_run_tools_worker_token(state: &AppState, run_id: &RunId) -> String {
     crate::worker_token::issue_worker_token_with_scopes(
-        &keys,
+        state.worker_token_keys(),
         run_id,
         crate::worker_token::WorkerScopeSet::run_worker_with_agent_run_tools(),
     )
@@ -846,12 +843,12 @@ async fn next_worker_control_frame(
 
 #[tokio::test(flavor = "current_thread")]
 async fn worker_control_stream_rejects_missing_user_and_cross_run_auth() {
-    let (_state, app) = jwt_auth_app();
+    let (state, app) = jwt_auth_app();
     let user_bearer = issue_test_user_jwt();
     let run_id = create_run_with_bearer(&app, &user_bearer).await;
-    let worker_bearer = issue_test_worker_token(&run_id);
+    let worker_bearer = issue_test_worker_token(&state, &run_id);
     let other_run_id = create_run_with_bearer(&app, &user_bearer).await;
-    let other_worker_bearer = issue_test_worker_token(&other_run_id);
+    let other_worker_bearer = issue_test_worker_token(&state, &other_run_id);
     let server = WorkerControlWsTestServer::spawn(app).await;
 
     assert_worker_control_ws_rejected(&server, run_id, None, None, StatusCode::UNAUTHORIZED).await;
@@ -883,7 +880,7 @@ async fn worker_control_stream_start_subscription_delivers_frames() {
     let (state, app) = jwt_auth_app();
     let user_bearer = issue_test_user_jwt();
     let run_id = create_run_with_bearer(&app, &user_bearer).await;
-    let worker_bearer = issue_test_worker_token(&run_id);
+    let worker_bearer = issue_test_worker_token(&state, &run_id);
     let server = WorkerControlWsTestServer::spawn(app).await;
     let mut socket = connect_worker_control_ws(&server, run_id, &worker_bearer, None).await;
 
@@ -904,7 +901,7 @@ async fn worker_control_stream_after_subscription_delivers_only_later_frames() {
     let (state, app) = jwt_auth_app();
     let user_bearer = issue_test_user_jwt();
     let run_id = create_run_with_bearer(&app, &user_bearer).await;
-    let worker_bearer = issue_test_worker_token(&run_id);
+    let worker_bearer = issue_test_worker_token(&state, &run_id);
     let first = state
         .worker_control_bus
         .publish(run_id, WorkerControlEnvelope::cancel_run())
@@ -934,7 +931,7 @@ async fn worker_control_stream_acknowledgements_settle_the_waiting_caller() {
     let (state, app) = jwt_auth_app();
     let user_bearer = issue_test_user_jwt();
     let run_id = create_run_with_bearer(&app, &user_bearer).await;
-    let worker_bearer = issue_test_worker_token(&run_id);
+    let worker_bearer = issue_test_worker_token(&state, &run_id);
     let server = WorkerControlWsTestServer::spawn(app).await;
     let mut socket = connect_worker_control_ws(&server, run_id, &worker_bearer, None).await;
 
@@ -973,10 +970,10 @@ async fn worker_control_stream_acknowledgements_settle_the_waiting_caller() {
 
 #[tokio::test(flavor = "current_thread")]
 async fn worker_control_stream_invalid_cursor_is_http_gone_before_upgrade() {
-    let (_state, app) = jwt_auth_app();
+    let (state, app) = jwt_auth_app();
     let user_bearer = issue_test_user_jwt();
     let run_id = create_run_with_bearer(&app, &user_bearer).await;
-    let worker_bearer = issue_test_worker_token(&run_id);
+    let worker_bearer = issue_test_worker_token(&state, &run_id);
     let server = WorkerControlWsTestServer::spawn(app).await;
 
     assert_worker_control_ws_rejected(
@@ -4108,7 +4105,7 @@ async fn run_tools_worker_cannot_select_server_folder_from_clone_based_parent()
     let (state, app) = jwt_auth_app();
     let user_token = issue_test_user_jwt();
     let parent_run_id = create_run_with_bearer(&app, &user_token).await;
-    let worker_token = issue_test_run_tools_worker_token(&parent_run_id);
+    let worker_token = issue_test_run_tools_worker_token(&state, &parent_run_id);
     let workflow_version_id = store_workflow_version(&state, MINIMAL_DOT, None).await;
     let mut intent = folder_intent(workflow_version_id, missing_target.to_string_lossy());
     intent["environment_id"] = json!("local");
@@ -4147,7 +4144,7 @@ async fn run_tools_worker_cannot_select_server_folder_from_clone_based_parent()
 async fn run_tools_worker_folder_target_from_missing_parent_run_is_not_found() {
     let dir = tempfile::tempdir().unwrap();
     let (state, app) = jwt_auth_app();
-    let worker_token = issue_test_run_tools_worker_token(&RunId::new());
+    let worker_token = issue_test_run_tools_worker_token(&state, &RunId::new());
     let workflow_version_id = store_workflow_version(&state, MINIMAL_DOT, None).await;
     let mut intent = folder_intent(workflow_version_id, dir.path().to_string_lossy());
     intent["environment_id"] = json!("local");
@@ -4196,7 +4193,7 @@ async fn run_tools_worker_can_select_server_folder_from_local_parent() {
         .unwrap();
     let parent = response_json!(response, StatusCode::CREATED).await;
     let parent_run_id = parent["id"].as_str().unwrap().parse::<RunId>().unwrap();
-    let worker_token = issue_test_run_tools_worker_token(&parent_run_id);
+    let worker_token = issue_test_run_tools_worker_token(&state, &parent_run_id);
     let mut child_intent = folder_intent(workflow_version_id, dir.path().to_string_lossy());
     child_intent["environment_id"] = json!("local");
     child_intent["parent_id"] = json!(parent_run_id);
@@ -7316,9 +7313,9 @@ async fn worker_token_accepts_run_scoped_routes_and_falls_back_to_user_jwt() {
     let (state, app) = jwt_auth_app();
     let user_jwt = issue_test_user_jwt();
     let run_id = create_run_with_bearer(&app, &user_jwt).await;
-    let worker_token = issue_test_worker_token(&run_id);
+    let worker_token = issue_test_worker_token(&state, &run_id);
     let other_run_id = create_run_with_bearer(&app, &user_jwt).await;
-    let other_worker_token = issue_test_worker_token(&other_run_id);
+    let other_worker_token = issue_test_worker_token(&state, &other_run_id);
     let blob_hash = state
         .store_ref()
         .blobs()
@@ -7422,7 +7419,7 @@ async fn run_tool_worker_token_can_use_client_backend_routes_across_runs() {
     let user_jwt = issue_test_user_jwt();
     let parent_run_id = create_run_with_bearer(&app, &user_jwt).await;
     let target_run_id = create_run_with_bearer(&app, &user_jwt).await;
-    let run_tool_worker_token = issue_test_run_tools_worker_token(&parent_run_id);
+    let run_tool_worker_token = issue_test_run_tools_worker_token(&state, &parent_run_id);
 
     let response = app
         .clone()
@@ -7575,11 +7572,11 @@ async fn run_tool_worker_token_can_use_client_backend_routes_across_runs() {
 
 #[tokio::test]
 async fn cross_run_base_worker_remains_forbidden_from_pair_routes() {
-    let (_state, app) = jwt_auth_app();
+    let (state, app) = jwt_auth_app();
     let user_jwt = issue_test_user_jwt();
     let origin_run_id = create_run_with_bearer(&app, &user_jwt).await;
     let target_run_id = create_run_with_bearer(&app, &user_jwt).await;
-    let worker_token = issue_test_worker_token(&origin_run_id);
+    let worker_token = issue_test_worker_token(&state, &origin_run_id);
 
     let response = app
         .clone()
@@ -7596,11 +7593,11 @@ async fn cross_run_base_worker_remains_forbidden_from_pair_routes() {
 
 #[tokio::test]
 async fn run_tools_worker_cannot_call_user_only_non_mcp_routes() {
-    let (_state, app) = jwt_auth_app();
+    let (state, app) = jwt_auth_app();
     let user_jwt = issue_test_user_jwt();
     let origin_run_id = create_run_with_bearer(&app, &user_jwt).await;
     let target_run_id = create_run_with_bearer(&app, &user_jwt).await;
-    let worker_token = issue_test_run_tools_worker_token(&origin_run_id);
+    let worker_token = issue_test_run_tools_worker_token(&state, &origin_run_id);
 
     for (method, path) in [
         (Method::POST, format!("/runs/{target_run_id}/approve")),
@@ -7629,10 +7626,10 @@ async fn run_tools_worker_cannot_call_user_only_non_mcp_routes() {
 
 #[tokio::test]
 async fn base_worker_token_is_rejected_by_run_tool_only_routes() {
-    let (_state, app) = jwt_auth_app();
+    let (state, app) = jwt_auth_app();
     let user_jwt = issue_test_user_jwt();
     let run_id = create_run_with_bearer(&app, &user_jwt).await;
-    let worker_token = issue_test_worker_token(&run_id);
+    let worker_token = issue_test_worker_token(&state, &run_id);
 
     for (method, path) in [
         (Method::GET, "/runs".to_string()),
@@ -7662,10 +7659,10 @@ async fn base_worker_token_is_rejected_by_run_tool_only_routes() {
 
 #[tokio::test]
 async fn worker_token_is_rejected_on_user_only_routes() {
-    let (_state, app) = jwt_auth_app();
+    let (state, app) = jwt_auth_app();
     let user_jwt = issue_test_user_jwt();
     let run_id = create_run_with_bearer(&app, &user_jwt).await;
-    let worker_token = issue_test_worker_token(&run_id);
+    let worker_token = issue_test_worker_token(&state, &run_id);
     let blob_hash = BlobHash::new(b"blob");
     let user_only_routes = vec![
         (Method::GET, "/runs".to_string()),
@@ -7832,7 +7829,7 @@ async fn worker_started_child_run_requires_approval_before_becoming_runnable() {
     let (state, app) = jwt_auth_app();
     let user_jwt = issue_test_user_jwt();
     let parent_run_id = create_run_with_bearer(&app, &user_jwt).await;
-    let worker_token = issue_test_run_tools_worker_token(&parent_run_id);
+    let worker_token = issue_test_run_tools_worker_token(&state, &parent_run_id);
     let mut child_intent =
         test_intent_with_bearer(&app, "workflow.fabro", MINIMAL_DOT, None, Some(&user_jwt)).await;
     child_intent["parent_id"] = json!(parent_run_id.to_string());
@@ -7931,10 +7928,10 @@ async fn worker_started_child_run_requires_approval_before_becoming_runnable() {
 
 #[tokio::test]
 async fn denying_pending_child_run_fails_with_approval_denied() {
-    let (_state, app) = jwt_auth_app();
+    let (state, app) = jwt_auth_app();
     let user_jwt = issue_test_user_jwt();
     let parent_run_id = create_run_with_bearer(&app, &user_jwt).await;
-    let worker_token = issue_test_run_tools_worker_token(&parent_run_id);
+    let worker_token = issue_test_run_tools_worker_token(&state, &parent_run_id);
     let mut child_intent =
         test_intent_with_bearer(&app, "workflow.fabro", MINIMAL_DOT, None, Some(&user_jwt)).await;
     child_intent["parent_id"] = json!(parent_run_id.to_string());
@@ -8947,10 +8944,10 @@ fn batch_delete_body(run_ids: &[RunId], force: bool) -> serde_json::Value {
 
 #[tokio::test]
 async fn batch_lifecycle_requires_user_authentication() {
-    let (_state, app) = jwt_auth_app();
+    let (state, app) = jwt_auth_app();
     let user_jwt = issue_test_user_jwt();
     let run_id = create_run_with_bearer(&app, &user_jwt).await;
-    let worker_token = issue_test_worker_token(&run_id);
+    let worker_token = issue_test_worker_token(&state, &run_id);
     let body = batch_lifecycle_body(&[run_id]);
 
     let unauthenticated = app
@@ -8984,10 +8981,10 @@ async fn batch_lifecycle_requires_user_authentication() {
 
 #[tokio::test]
 async fn batch_delete_requires_user_authentication() {
-    let (_state, app) = jwt_auth_app();
+    let (state, app) = jwt_auth_app();
     let user_jwt = issue_test_user_jwt();
     let run_id = create_run_with_bearer(&app, &user_jwt).await;
-    let worker_token = issue_test_worker_token(&run_id);
+    let worker_token = issue_test_worker_token(&state, &run_id);
     let body = batch_delete_body(&[run_id], false);
 
     let unauthenticated = app
@@ -10637,10 +10634,13 @@ async fn workflow_version_registration_requires_user_or_run_tools_capability() {
     for (token, expected) in [
         (issue_test_user_jwt(), StatusCode::CREATED),
         (
-            issue_test_run_tools_worker_token(&run_id),
+            issue_test_run_tools_worker_token(&state, &run_id),
             StatusCode::CREATED,
         ),
-        (issue_test_worker_token(&run_id), StatusCode::FORBIDDEN),
+        (
+            issue_test_worker_token(&state, &run_id),
+            StatusCode::FORBIDDEN,
+        ),
     ] {
         let response = app
             .clone()
@@ -10805,7 +10805,7 @@ fn slack_service_respects_disabled_server_config_even_with_vault_tokens() {
 async fn run_tools_worker_registers_contents_then_creates_by_version_id() {
     let (state, app) = jwt_auth_app();
     let parent_id = create_run_with_bearer(&app, &issue_test_user_jwt()).await;
-    let token = issue_test_run_tools_worker_token(&parent_id);
+    let token = issue_test_run_tools_worker_token(&state, &parent_id);
     let response = app
         .clone()
         .oneshot(json_bearer_request(

lib/apps/fabro-server/src/server/tests/artifact_storage.rs

@@ -25,7 +25,7 @@ async fn artifact_upload_accepts_large_and_exact_limit_content_without_sqlite_or
     let (state, app) = jwt_auth_app();
     let user = issue_test_user_jwt();
     let run = create_run_with_bearer(&app, &user).await;
-    let worker = issue_test_worker_token(&run);
+    let worker = issue_test_worker_token(&state, &run);
     for size in [3 * 1024 * 1024, ARTIFACT_MAX_FILE_BYTES] {
         let bytes = vec![0xa3; size];
         let hash = BlobHash::new(&bytes);
@@ -87,8 +87,8 @@ async fn artifact_upload_rejects_unauthorized_invalid_and_oversized_bodies_witho
     let (state, app) = jwt_auth_app();
     let user = issue_test_user_jwt();
     let run = create_run_with_bearer(&app, &user).await;
-    let worker = issue_test_worker_token(&run);
-    let other_worker = issue_test_worker_token(&RunId::new());
+    let worker = issue_test_worker_token(&state, &run);
+    let other_worker = issue_test_worker_token(&state, &RunId::new());
     let hash = BlobHash::new(b"valid content");
     state
         .artifact_store
@@ -152,7 +152,7 @@ async fn artifact_upload_rejects_unauthorized_invalid_and_oversized_bodies_witho
         .oneshot(upload(
             missing_run,
             &digest,
-            &issue_test_worker_token(&missing_run),
+            &issue_test_worker_token(&state, &missing_run),
             Body::from("valid content"),
         ))
         .await
@@ -246,7 +246,7 @@ async fn artifact_worker_client_uses_the_configured_s3_backend_and_prefix() {
         })
         .await;
     let server = WorkerControlWsTestServer::spawn(app).await;
-    let worker_token = issue_test_worker_token(&run);
+    let worker_token = issue_test_worker_token(&state, &run);
     let mut headers = fabro_http::header::HeaderMap::new();
     headers.insert(
         fabro_http::header::AUTHORIZATION,

lib/apps/fabro-server/src/test_support.rs

@@ -28,6 +28,7 @@ use fabro_types::{
 use fabro_vault::{SecretType, Vault};
 use lithos_llm::catalog::ProviderId;
 use object_store::memory::InMemory as MemoryObjectStore;
+use tokio::fs as tokio_fs;
 use tokio::runtime::Builder as TokioRuntimeBuilder;
 use tokio_util::sync::CancellationToken;
 use ulid::Ulid;
@@ -745,6 +746,20 @@ pub fn build_test_router(state: Arc<AppState>) -> Router {
     with_test_user(server::build_router(state, test_auth_mode()))
 }
 
+/// The CLI's direct-worker tests sign their own tokens with the isolated
+/// server's test secret. Publish only the non-secret audience so those
+/// tokens address the actual server instance, without overriding the
+/// production fence or persisting any worker credential.
+pub(crate) async fn write_test_worker_audience(state: &AppState) -> anyhow::Result<()> {
+    let path = Storage::new(state.server_storage_dir())
+        .runtime_directory()
+        .record_path()
+        .with_file_name("test-worker-audience");
+    tokio_fs::create_dir_all(path.parent().expect("the runtime path has a parent")).await?;
+    tokio_fs::write(path, state.worker_token_keys().test_audience()).await?;
+    Ok(())
+}
+
 pub fn build_test_router_with_options(state: Arc<AppState>, options: RouterOptions) -> Router {
     with_test_user(server::build_router_with_options(
         state,

lib/apps/fabro-server/src/worker_token.rs

@@ -22,20 +22,27 @@ pub(crate) struct WorkerTokenKeys {
     encoding:   Arc<EncodingKey>,
     decoding:   Arc<DecodingKey>,
     validation: Arc<Validation>,
+    /// Startup recovery replaces workers. Their credentials must therefore
+    /// be valid only for the server instance that launched them, including
+    /// platform writes made before acquiring or after releasing a lease.
+    audience:   String,
 }
 
 impl WorkerTokenKeys {
     pub(crate) fn from_master_secret(secret: &[u8]) -> Result<Self, KeyDeriveError> {
         let key = auth::derive_worker_jwt_key(secret)?;
+        let audience = Uuid::new_v4().to_string();
         let mut validation = Validation::new(Algorithm::HS256);
         validation.validate_nbf = false;
-        validation.set_required_spec_claims(&["iss", "iat", "exp"]);
+        validation.set_required_spec_claims(&["iss", "iat", "exp", "aud"]);
         validation.set_issuer(&[WORKER_TOKEN_ISSUER]);
+        validation.set_audience(&[&audience]);
 
         Ok(Self {
-            encoding:   Arc::new(EncodingKey::from_secret(&key)),
-            decoding:   Arc::new(DecodingKey::from_secret(&key)),
+            encoding: Arc::new(EncodingKey::from_secret(&key)),
+            decoding: Arc::new(DecodingKey::from_secret(&key)),
             validation: Arc::new(validation),
+            audience,
         })
     }
 
@@ -48,11 +55,17 @@ impl WorkerTokenKeys {
     pub(crate) fn validation(&self) -> &Validation {
         &self.validation
     }
+
+    #[cfg(any(test, feature = "test-support"))]
+    pub(crate) fn test_audience(&self) -> &str {
+        &self.audience
+    }
 }
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
 pub(crate) struct WorkerTokenClaims {
     pub(crate) iss:    String,
+    pub(crate) aud:    String,
     pub(crate) iat:    u64,
     pub(crate) exp:    u64,
     pub(crate) run_id: String,
@@ -117,6 +130,7 @@ pub(crate) fn issue_worker_token_with_scopes(
         .map_or(0, |duration| duration.as_secs());
     let claims = WorkerTokenClaims {
         iss:    WORKER_TOKEN_ISSUER.to_string(),
+        aud:    keys.audience.clone(),
         iat:    now,
         exp:    now + WORKER_TOKEN_TTL_SECS,
         run_id: run_id.to_string(),
@@ -220,6 +234,7 @@ mod tests {
     ) -> String {
         let claims = WorkerTokenClaims {
             iss:    WORKER_TOKEN_ISSUER.to_string(),
+            aud:    keys.audience.clone(),
             iat:    1,
             exp:    u64::MAX / 2,
             run_id: run_id.to_string(),
@@ -233,6 +248,7 @@ mod tests {
     fn expired_worker_token(keys: &WorkerTokenKeys, run_id: &fabro_types::RunId) -> String {
         let claims = WorkerTokenClaims {
             iss:    WORKER_TOKEN_ISSUER.to_string(),
+            aud:    keys.audience.clone(),
             iat:    1,
             exp:    2,
             run_id: run_id.to_string(),
@@ -276,6 +292,7 @@ mod tests {
 
         assert_eq!(decoded.claims, WorkerTokenClaims {
             iss:    WORKER_TOKEN_ISSUER.to_string(),
+            aud:    keys.audience.clone(),
             iat:    decoded.claims.iat,
             exp:    decoded.claims.exp,
             run_id: run_id.to_string(),
@@ -288,16 +305,49 @@ mod tests {
     }
 
     #[test]
-    fn worker_token_survives_key_rederivation() {
+    fn worker_token_is_rejected_by_a_restarted_server_with_the_same_secret() {
         let run_id = run_id();
         let first = keys(TEST_SECRET);
         let second = keys(TEST_SECRET);
 
         let token = issue_worker_token(&first, &run_id).expect("worker token should issue");
-        let decoded = decode::<WorkerTokenClaims>(&token, &second.decoding, &second.validation)
-            .expect("worker token should decode after re-derivation");
+        assert_eq!(
+            decode_worker_token(&token, &second),
+            Err(JwtError::AccessTokenInvalid)
+        );
+        let replacement = issue_worker_token(&second, &run_id).unwrap();
+        assert_eq!(
+            decode_worker_token(&replacement, &second).unwrap().run_id,
+            run_id
+        );
+    }
 
-        assert_eq!(decoded.claims.run_id, run_id.to_string());
+    #[test]
+    fn worker_token_is_valid_for_clones_of_the_issuing_server_keys() {
+        let keys = keys(TEST_SECRET);
+        let token = issue_worker_token(&keys, &run_id()).unwrap();
+        assert_eq!(
+            decode_worker_token(&token, &keys.clone()).unwrap().run_id,
+            run_id()
+        );
+    }
+
+    #[test]
+    fn worker_token_without_a_server_audience_is_rejected() {
+        let keys = keys(TEST_SECRET);
+        let token = issue_worker_token(&keys, &run_id()).unwrap();
+        let mut claims = serde_json::to_value(
+            decode::<WorkerTokenClaims>(&token, &keys.decoding, &keys.validation)
+                .unwrap()
+                .claims,
+        )
+        .unwrap();
+        claims.as_object_mut().unwrap().remove("aud");
+        let legacy = jsonwebtoken::encode(&worker_token_header(), &claims, &keys.encoding).unwrap();
+        assert_eq!(
+            decode_worker_token(&legacy, &keys),
+            Err(JwtError::AccessTokenInvalid)
+        );
     }
 
     #[test]