lash-core 0.1.0-alpha.83

//! [`EffectHost`] scope-factory and effect-controller replay conformance.

use super::*;

/// One scope selected by an [`EffectHost`] and one effect envelope executed
/// through the scoped controller.
#[derive(Clone, Debug)]
pub struct RecordingEffectHostRecord {
    pub runtime_scope: RuntimeScope,
    pub execution_scope: ExecutionScope,
    pub effect_id: String,
    pub effect_kind: RuntimeEffectKind,
    pub replay_key: Option<String>,
    pub envelope_hash: String,
}

#[derive(Clone)]
struct RecordingEffectHostController {
    execution_scope: ExecutionScope,
    records: Arc<Mutex<Vec<RecordingEffectHostRecord>>>,
}

impl crate::AwaitEventResolver for RecordingEffectHostController {}

#[async_trait::async_trait]
impl RuntimeEffectController for RecordingEffectHostController {
    async fn execute_effect(
        &self,
        envelope: RuntimeEffectEnvelope,
        _local_executor: RuntimeEffectLocalExecutor<'_>,
    ) -> Result<RuntimeEffectOutcome, RuntimeEffectControllerError> {
        let envelope_hash = envelope.stable_hash()?;
        self.records
            .lock()
            .expect("effect host records")
            .push(RecordingEffectHostRecord {
                runtime_scope: envelope.invocation.scope.clone(),
                execution_scope: self.execution_scope.clone(),
                effect_id: envelope
                    .invocation
                    .effect_id()
                    .expect("effect invocation")
                    .to_string(),
                effect_kind: envelope.command.kind(),
                replay_key: envelope.invocation.replay_key().map(ToOwned::to_owned),
                envelope_hash,
            });
        match envelope.command {
            RuntimeEffectCommand::Sleep { .. } => Ok(RuntimeEffectOutcome::Sleep),
            RuntimeEffectCommand::DurableStep { input, .. } => {
                Ok(RuntimeEffectOutcome::DurableStep { value: input })
            }
            command => Err(RuntimeEffectControllerError::new(
                "recording_effect_host_unsupported_command",
                format!(
                    "recording effect host cannot synthesize {} outcomes",
                    command.kind().as_str()
                ),
            )),
        }
    }
}

/// Test fixture that records every selected [`ExecutionScope`] and every effect
/// envelope executed through the returned scoped controller.
#[derive(Clone, Default)]
pub struct RecordingEffectHost {
    selected_scopes: Arc<Mutex<Vec<ExecutionScope>>>,
    records: Arc<Mutex<Vec<RecordingEffectHostRecord>>>,
}

impl RecordingEffectHost {
    pub fn selected_scopes(&self) -> Vec<ExecutionScope> {
        self.selected_scopes
            .lock()
            .expect("selected execution scopes")
            .clone()
    }

    pub fn records(&self) -> Vec<RecordingEffectHostRecord> {
        self.records.lock().expect("effect host records").clone()
    }

    fn scoped_for<'run>(
        &self,
        scope: ExecutionScope,
    ) -> Result<ScopedEffectController<'run>, crate::RuntimeError> {
        self.selected_scopes
            .lock()
            .expect("selected execution scopes")
            .push(scope.clone());
        ScopedEffectController::shared(
            Arc::new(RecordingEffectHostController {
                execution_scope: scope.clone(),
                records: Arc::clone(&self.records),
            }),
            scope,
        )
    }
}

impl crate::AwaitEventResolver for RecordingEffectHost {}

impl EffectHost for RecordingEffectHost {
    fn scoped<'run>(
        &'run self,
        scope: ExecutionScope,
    ) -> Result<ScopedEffectController<'run>, crate::RuntimeError> {
        self.scoped_for(scope)
    }

    fn scoped_static(
        &self,
        scope: ExecutionScope,
    ) -> Result<Option<ScopedEffectController<'static>>, crate::RuntimeError> {
        Ok(Some(self.scoped_for(scope)?))
    }
}

/// Run the generic [`EffectHost`] scope-factory conformance suite.
///
/// This suite checks the deployment-level contract: execution scopes must carry
/// stable semantic identity, empty ids must fail loudly, and hosts that expose
/// a static scoped controller must preserve the same scope metadata. It does
/// not assert durability; that remains a property of each implementation.
/// Substrate-native hosts such as Restate complete the in-flight contract at
/// this effect-host/controller boundary; [`RuntimePersistence`] remains the
/// committed-state store contract, not a workflow-history contract.
pub async fn effect_host<F>(make: F)
where
    F: Fn() -> Arc<dyn EffectHost>,
{
    effect_host_preserves_scope_metadata(make()).await;
    effect_host_rejects_missing_scope_ids(make()).await;
    effect_host_static_scope_preserves_metadata_when_available(make()).await;
}

/// Run the generic AwaitEvent conformance suite for hosts that implement the
/// external completion primitive.
///
/// This is intentionally separate from [`effect_host`]: deployment-level hosts
/// may be valid scope factories while requiring an external workflow/object
/// context before an AwaitEvent can be awaited.
pub async fn effect_host_await_events<F>(make: F)
where
    F: Fn() -> Arc<dyn EffectHost>,
{
    effect_host_await_event_key_is_stable(make()).await;
    effect_host_await_event_accepts_early_resolution(make()).await;
    effect_host_await_event_duplicate_resolution_is_terminal(make()).await;
    effect_host_await_event_cancel_and_timeout_are_terminal(make()).await;
    effect_host_await_event_revokes_session_scope(make()).await;
    effect_host_await_event_session_cancel_resolves_outstanding_waits(make()).await;
    effect_host_await_event_rejects_tampered_keys(make()).await;
}

/// Run the generic durable-step conformance suite for effect hosts that expose
/// durable tool effects through their scoped controllers.
#[cfg(any(test, feature = "testing"))]
pub async fn effect_host_durable_steps<F>(make: F)
where
    F: Fn() -> Arc<dyn EffectHost>,
{
    let host = make();
    assert!(
        host.supports_durable_effects(),
        "durable-step conformance requires a host that advertises durable effects"
    );
    let scoped = host
        .scoped(ExecutionScope::turn(
            "durable-step-session",
            "durable-step-turn",
        ))
        .expect("durable-step scope");
    effect_controller_durable_steps(scoped.controller()).await;
}

/// Run durable-step execution checks for a handler-scoped controller.
#[cfg(any(test, feature = "testing"))]
pub async fn effect_controller_durable_steps(controller: &dyn RuntimeEffectController) {
    assert!(
        controller.supports_durable_effects(),
        "durable-step conformance requires a controller that advertises durable effects"
    );
    let first = durable_step_conformance_envelope("create", serde_json::json!({ "x": 1 }));
    let second = durable_step_conformance_envelope("create", serde_json::json!({ "x": 1 }));
    assert_eq!(
        first.stable_hash().expect("first hash"),
        second.stable_hash().expect("second hash"),
        "same step id/input/replay key must hash stably"
    );
    assert!(
        first.invocation.replay_key().is_some(),
        "durable-step envelope must carry replay.key"
    );
    let changed = durable_step_conformance_envelope("create", serde_json::json!({ "x": 2 }));
    assert_ne!(
        first.stable_hash().expect("first hash"),
        changed.stable_hash().expect("changed hash"),
        "durable-step input must participate in the stable envelope hash"
    );

    let outcome = controller
        .execute_effect(
            first,
            RuntimeEffectLocalExecutor::durable_step(|input| async move {
                Ok(serde_json::json!({ "recorded": input }))
            }),
        )
        .await
        .expect("durable-step success");
    assert_eq!(
        outcome.into_durable_step().expect("durable-step outcome"),
        serde_json::json!({ "recorded": { "x": 1 } })
    );

    let err = controller
        .execute_effect(
            durable_step_conformance_envelope("error", serde_json::json!({ "x": 1 })),
            RuntimeEffectLocalExecutor::durable_step(|_| async {
                Err(crate::RuntimeError::new(
                    "durable_step_conformance_error",
                    "durable step failed",
                ))
            }),
        )
        .await
        .expect_err("durable-step error must be recorded as controller error");
    assert_eq!(err.code, "durable_step_conformance_error");
}

/// Run durable-step replay checks for controllers with an explicit replay mode.
#[cfg(any(test, feature = "testing"))]
pub async fn effect_controller_durable_steps_replay(
    controller: &dyn RuntimeEffectController,
    start_replay: impl FnOnce(),
) {
    effect_controller_durable_steps(controller).await;
    let success =
        durable_step_conformance_envelope("replay-success", serde_json::json!({ "n": 1 }));
    let error = durable_step_conformance_envelope("replay-error", serde_json::json!({ "n": 2 }));

    let first_success = controller
        .execute_effect(
            success.clone(),
            RuntimeEffectLocalExecutor::durable_step(|input| async move {
                Ok(serde_json::json!({ "first": input }))
            }),
        )
        .await
        .expect("first durable-step success");
    assert_eq!(
        first_success
            .into_durable_step()
            .expect("first durable-step value"),
        serde_json::json!({ "first": { "n": 1 } })
    );
    let first_error = controller
        .execute_effect(
            error.clone(),
            RuntimeEffectLocalExecutor::durable_step(|_| async {
                Err(crate::RuntimeError::new(
                    "durable_step_replay_error",
                    "recorded error",
                ))
            }),
        )
        .await
        .expect_err("first durable-step error");
    assert_eq!(first_error.code, "durable_step_replay_error");

    start_replay();
    let local_calls = Arc::new(Mutex::new(Vec::new()));
    let replay_success = controller
        .execute_effect(
            success,
            durable_step_conformance_failing_executor(Arc::clone(&local_calls)),
        )
        .await
        .expect("replayed durable-step success");
    assert_eq!(
        replay_success
            .into_durable_step()
            .expect("replayed durable-step value"),
        serde_json::json!({ "first": { "n": 1 } })
    );
    let replay_error = controller
        .execute_effect(
            error,
            durable_step_conformance_failing_executor(Arc::clone(&local_calls)),
        )
        .await
        .expect_err("replayed durable-step error");
    assert_eq!(replay_error.code, "durable_step_replay_error");
    assert!(
        local_calls.lock().expect("local calls").is_empty(),
        "durable-step replay must not invoke local closures"
    );
}

async fn effect_host_preserves_scope_metadata(host: Arc<dyn EffectHost>) {
    let scope = ExecutionScope::queue_drain("session-1", "drain-1");
    let scoped = host.scoped(scope.clone()).expect("queue drain scope");
    assert_eq!(
        scoped.execution_scope(),
        &scope,
        "scoped controller must retain the selected semantic scope"
    );
    assert_eq!(scoped.scope_id(), "drain-1");
    assert_eq!(scoped.turn_id(), None);

    let turn_scope = ExecutionScope::turn("session-1", "turn-1");
    let scoped_turn = host.scoped(turn_scope.clone()).expect("turn scope");
    assert_eq!(scoped_turn.execution_scope(), &turn_scope);
    assert_eq!(scoped_turn.scope_id(), "turn-1");
    assert_eq!(scoped_turn.turn_id(), Some("turn-1"));
}

async fn effect_host_rejects_missing_scope_ids(host: Arc<dyn EffectHost>) {
    let invalid_scopes = [
        ExecutionScope::turn("", "turn"),
        ExecutionScope::turn("session", ""),
        ExecutionScope::process(""),
        ExecutionScope::queue_drain("session", ""),
        ExecutionScope::session_delete(""),
        ExecutionScope::runtime_operation(""),
    ];

    for scope in invalid_scopes {
        let err = match host.scoped(scope) {
            Ok(_) => panic!("invalid execution scope must be rejected"),
            Err(err) => err,
        };
        assert_eq!(
            err.code,
            crate::RuntimeErrorCode::MissingExecutionScopeId,
            "invalid scope ids must fail with the stable missing-scope code"
        );
    }
}

async fn effect_host_static_scope_preserves_metadata_when_available(host: Arc<dyn EffectHost>) {
    let scope = ExecutionScope::runtime_operation("static-runtime-op");
    let Some(scoped) = host
        .scoped_static(scope.clone())
        .expect("static scope factory")
    else {
        return;
    };
    assert_eq!(scoped.execution_scope(), &scope);
    assert_eq!(scoped.scope_id(), "static-runtime-op");
}

async fn effect_host_await_event_key_is_stable(host: Arc<dyn EffectHost>) {
    let scope = ExecutionScope::turn("await-event-session-stable", "turn-stable");
    let wait = AwaitEventWaitIdentity::tool_completion("call-stable");

    let first = host
        .await_event_key(&scope, wait.clone())
        .await
        .expect("first await-event key");
    let second = host
        .await_event_key(&scope, wait)
        .await
        .expect("second await-event key");

    assert_eq!(first, second);
}

async fn effect_host_await_event_accepts_early_resolution(host: Arc<dyn EffectHost>) {
    let scope = ExecutionScope::turn("await-event-session-early", "turn-early");
    let key = host
        .await_event_key(
            &scope,
            AwaitEventWaitIdentity::tool_completion("call-early"),
        )
        .await
        .expect("await-event key");
    let resolution = Resolution::Ok(serde_json::json!({ "ready": true }));

    assert_eq!(
        host.resolve_await_event(&key, resolution.clone())
            .await
            .expect("early resolve"),
        ResolveOutcome::Accepted
    );
    let awaited = host
        .await_await_event(&key, tokio_util::sync::CancellationToken::new(), None)
        .await
        .expect("await early-resolved event");
    assert_eq!(awaited, resolution);
}

async fn effect_host_await_event_duplicate_resolution_is_terminal(host: Arc<dyn EffectHost>) {
    let scope = ExecutionScope::turn("await-event-session-dupe", "turn-dupe");
    let key = host
        .await_event_key(&scope, AwaitEventWaitIdentity::tool_completion("call-dupe"))
        .await
        .expect("await-event key");
    let resolution = Resolution::Ok(serde_json::json!("first"));

    let first = host
        .resolve_await_event(&key, resolution.clone())
        .await
        .expect("first resolve");
    let second = host
        .resolve_await_event(&key, Resolution::Ok(serde_json::json!("second")))
        .await
        .expect("duplicate resolve");

    assert_eq!(first, ResolveOutcome::Accepted);
    assert_eq!(
        second,
        ResolveOutcome::AlreadyResolved {
            terminal: resolution
        }
    );
}

async fn effect_host_await_event_cancel_and_timeout_are_terminal(host: Arc<dyn EffectHost>) {
    let cancel_scope = ExecutionScope::turn("await-event-session-cancel", "turn-cancel");
    let cancel_key = host
        .await_event_key(
            &cancel_scope,
            AwaitEventWaitIdentity::tool_completion("call-cancel"),
        )
        .await
        .expect("cancel await-event key");
    let cancel = tokio_util::sync::CancellationToken::new();
    cancel.cancel();
    let cancelled = host
        .await_await_event(&cancel_key, cancel, None)
        .await
        .expect("cancelled await-event");
    assert_eq!(cancelled, Resolution::Cancelled);
    assert_eq!(
        host.resolve_await_event(&cancel_key, Resolution::Ok(serde_json::json!("late")))
            .await
            .expect("late cancel resolve"),
        ResolveOutcome::AlreadyResolved {
            terminal: Resolution::Cancelled
        }
    );

    let timeout_scope = ExecutionScope::turn("await-event-session-timeout", "turn-timeout");
    let timeout_key = host
        .await_event_key(
            &timeout_scope,
            AwaitEventWaitIdentity::tool_completion("call-timeout"),
        )
        .await
        .expect("timeout await-event key");
    let timed_out = host
        .await_await_event(
            &timeout_key,
            tokio_util::sync::CancellationToken::new(),
            Some(std::time::Instant::now()),
        )
        .await
        .expect("timed-out await-event");
    assert_eq!(timed_out, Resolution::Timeout);
}

async fn effect_host_await_event_revokes_session_scope(host: Arc<dyn EffectHost>) {
    let scope = ExecutionScope::turn("await-event-session-revoke", "turn-revoke");
    let key = host
        .await_event_key(
            &scope,
            AwaitEventWaitIdentity::tool_completion("call-revoke"),
        )
        .await
        .expect("await-event key");

    host.revoke_await_events_for_session("await-event-session-revoke")
        .await
        .expect("revoke session");

    assert_eq!(
        host.resolve_await_event(&key, Resolution::Ok(serde_json::json!("late")))
            .await
            .expect("resolve revoked key"),
        ResolveOutcome::UnknownOrRevoked
    );
    let err = host
        .await_await_event(&key, tokio_util::sync::CancellationToken::new(), None)
        .await
        .expect_err("revoked key must not await");
    assert_eq!(err.code.as_str(), "await_event_unknown_or_revoked");
}

/// The standalone wait-revocation lever: cancelling a session's durable waits
/// resolves every *outstanding* wait with [`Resolution::Cancelled`] (waiters
/// never hang; late resolves observe the terminal) while leaving the session
/// usable — new waits registered afterwards resolve normally, unlike the
/// tombstoning session revocation exercised above.
async fn effect_host_await_event_session_cancel_resolves_outstanding_waits(
    host: Arc<dyn EffectHost>,
) {
    let scope = ExecutionScope::turn("await-event-session-cancel-waits", "turn-cancel-waits");
    let key = host
        .await_event_key(
            &scope,
            AwaitEventWaitIdentity::tool_completion("call-cancel-waits"),
        )
        .await
        .expect("await-event key");

    let waiter_host = Arc::clone(&host);
    let waiter_key = key.clone();
    let waiter = tokio::spawn(async move {
        waiter_host
            .await_await_event(
                &waiter_key,
                tokio_util::sync::CancellationToken::new(),
                None,
            )
            .await
    });
    // The spawned waiter registers its wait asynchronously; cancel repeatedly
    // (the lever is idempotent) until the waiter observes a terminal.
    let waited = tokio::time::timeout(std::time::Duration::from_secs(5), async {
        loop {
            host.cancel_await_events_for_session("await-event-session-cancel-waits")
                .await
                .expect("cancel session waits");
            if waiter.is_finished() {
                return waiter.await;
            }
            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
        }
    })
    .await
    .expect("outstanding wait must terminate after session cancel")
    .expect("waiter task joins")
    .expect("cancelled wait resolves rather than erroring");
    assert_eq!(
        waited,
        Resolution::Cancelled,
        "outstanding waits must resolve with the Cancelled terminal"
    );
    assert_eq!(
        host.resolve_await_event(&key, Resolution::Ok(serde_json::json!("late")))
            .await
            .expect("late resolve after cancel"),
        ResolveOutcome::AlreadyResolved {
            terminal: Resolution::Cancelled
        }
    );

    // The session is NOT tombstoned: a wait registered after the cancel still
    // resolves normally.
    let later_key = host
        .await_event_key(
            &scope,
            AwaitEventWaitIdentity::tool_completion("call-after-cancel"),
        )
        .await
        .expect("post-cancel await-event key");
    assert_eq!(
        host.resolve_await_event(&later_key, Resolution::Ok(serde_json::json!("still-works")))
            .await
            .expect("post-cancel resolve"),
        ResolveOutcome::Accepted
    );
    assert_eq!(
        host.await_await_event(&later_key, tokio_util::sync::CancellationToken::new(), None)
            .await
            .expect("post-cancel wait resolves"),
        Resolution::Ok(serde_json::json!("still-works"))
    );
}

async fn effect_host_await_event_rejects_tampered_keys(host: Arc<dyn EffectHost>) {
    let scope = ExecutionScope::turn("await-event-session-tamper", "turn-tamper");
    let mut key = host
        .await_event_key(
            &scope,
            AwaitEventWaitIdentity::tool_completion("call-tamper"),
        )
        .await
        .expect("await-event key");
    key.signature.push_str("-tampered");

    assert_eq!(
        host.resolve_await_event(&key, Resolution::Ok(serde_json::json!("bad")))
            .await
            .expect("resolve tampered key"),
        ResolveOutcome::UnknownOrRevoked
    );
    let err = host
        .await_await_event(&key, tokio_util::sync::CancellationToken::new(), None)
        .await
        .expect_err("tampered key must not await");
    assert_eq!(err.code.as_str(), "await_event_unknown_or_revoked");
}

/// Run the concurrent recorded-effect replay conformance case for a
/// handler-scoped durable controller.
///
/// The first pass starts two recorded effects concurrently and intentionally
/// lets the second finish before the first. After `start_replay`, the same
/// effects are requested in the opposite order with local executors that fail
/// if called. A compliant controller returns the recorded outcomes by
/// `replay.key`, independent of local completion/request ordering.
#[cfg(any(test, feature = "testing"))]
pub async fn effect_controller_concurrent_replay_deterministic(
    controller: &dyn RuntimeEffectController,
    start_replay: impl FnOnce(),
) {
    let slow = replay_conformance_exec_envelope("effect-slow");
    let fast = replay_conformance_exec_envelope("effect-fast");
    let completion_order = Arc::new(Mutex::new(Vec::new()));
    let barrier = Arc::new(tokio::sync::Barrier::new(2));
    let release_slow = Arc::new(tokio::sync::Notify::new());

    let first_pass = tokio::time::timeout(std::time::Duration::from_secs(2), async {
        tokio::join!(
            controller.execute_effect(
                slow.clone(),
                replay_conformance_recording_executor(
                    "effect-slow",
                    Arc::clone(&barrier),
                    Arc::clone(&release_slow),
                    Arc::clone(&completion_order),
                ),
            ),
            controller.execute_effect(
                fast.clone(),
                replay_conformance_recording_executor(
                    "effect-fast",
                    Arc::clone(&barrier),
                    Arc::clone(&release_slow),
                    Arc::clone(&completion_order),
                ),
            ),
        )
    })
    .await
    .expect("concurrent first-pass effects must both enter their local executors");
    let slow_first = first_pass.0.expect("slow first pass");
    let fast_first = first_pass.1.expect("fast first pass");
    assert_replay_conformance_exec_marker(slow_first, "effect-slow");
    assert_replay_conformance_exec_marker(fast_first, "effect-fast");
    assert_eq!(
        completion_order
            .lock()
            .expect("completion order")
            .as_slice(),
        &["effect-fast".to_string(), "effect-slow".to_string()],
        "first pass must prove local completion order can differ from effect request order"
    );

    start_replay();
    let replay_local_calls = Arc::new(Mutex::new(Vec::new()));
    let replay_pass = tokio::time::timeout(std::time::Duration::from_secs(2), async {
        tokio::join!(
            controller.execute_effect(
                fast,
                replay_conformance_failing_executor(Arc::clone(&replay_local_calls)),
            ),
            controller.execute_effect(
                slow,
                replay_conformance_failing_executor(Arc::clone(&replay_local_calls)),
            ),
        )
    })
    .await
    .expect("concurrent replay effects must resolve from host history");
    let fast_replay = replay_pass.0.expect("fast replay");
    let slow_replay = replay_pass.1.expect("slow replay");
    assert_replay_conformance_exec_marker(fast_replay, "effect-fast");
    assert_replay_conformance_exec_marker(slow_replay, "effect-slow");
    assert!(
        replay_local_calls
            .lock()
            .expect("replay local calls")
            .is_empty(),
        "replay must return recorded outcomes without invoking local executors"
    );
}

/// Run the tool-attempt replay conformance case for a handler-scoped durable
/// controller.
///
/// Store-backed controllers that support overlapping effect calls record two
/// child attempts concurrently and replay them in reverse order. Ordered
/// workflow-context controllers record them sequentially, then still replay in
/// reverse order. In both modes, outcomes must resolve by stable `replay.key`
/// rather than request position, completion order, or source order.
#[cfg(any(test, feature = "testing"))]
pub async fn effect_controller_tool_attempt_fanout_replay_deterministic(
    controller: &dyn RuntimeEffectController,
    start_replay: impl FnOnce(),
) {
    let slow =
        replay_conformance_tool_attempt_envelope("tool-attempt-slow", "call-slow", "slow_tool");
    let fast =
        replay_conformance_tool_attempt_envelope("tool-attempt-fast", "call-fast", "fast_tool");
    let completion_order = Arc::new(Mutex::new(Vec::new()));

    let first_pass = if controller.supports_concurrent_effects() {
        let barrier = Arc::new(tokio::sync::Barrier::new(2));
        let release_slow = Arc::new(tokio::sync::Notify::new());
        let first_pass = tokio::time::timeout(std::time::Duration::from_secs(2), async {
            tokio::join!(
                controller.execute_effect(
                    slow.clone(),
                    replay_conformance_tool_attempt_recording_executor(
                        "tool-attempt-slow",
                        "call-slow",
                        "slow_tool",
                        Some((
                            Arc::clone(&barrier),
                            Arc::clone(&release_slow),
                            Arc::clone(&completion_order),
                        )),
                    ),
                ),
                controller.execute_effect(
                    fast.clone(),
                    replay_conformance_tool_attempt_recording_executor(
                        "tool-attempt-fast",
                        "call-fast",
                        "fast_tool",
                        Some((
                            Arc::clone(&barrier),
                            Arc::clone(&release_slow),
                            Arc::clone(&completion_order),
                        )),
                    ),
                ),
            )
        })
        .await
        .expect("concurrent tool-attempt first pass must enter both local executors");
        assert_eq!(
            completion_order
                .lock()
                .expect("tool-attempt completion order")
                .as_slice(),
            &[
                "tool-attempt-fast".to_string(),
                "tool-attempt-slow".to_string()
            ],
            "first pass must prove tool-attempt completion order can differ from source order"
        );
        first_pass
    } else {
        let slow_first = controller
            .execute_effect(
                slow.clone(),
                replay_conformance_tool_attempt_recording_executor(
                    "tool-attempt-slow",
                    "call-slow",
                    "slow_tool",
                    None,
                ),
            )
            .await;
        let fast_first = controller
            .execute_effect(
                fast.clone(),
                replay_conformance_tool_attempt_recording_executor(
                    "tool-attempt-fast",
                    "call-fast",
                    "fast_tool",
                    None,
                ),
            )
            .await;
        assert_eq!(
            completion_order
                .lock()
                .expect("tool-attempt completion order")
                .as_slice(),
            &[] as &[String],
            "serial conformance path must not use the concurrent completion-order probe"
        );
        (slow_first, fast_first)
    };

    let slow_first = first_pass.0.expect("slow tool-attempt first pass");
    let fast_first = first_pass.1.expect("fast tool-attempt first pass");
    assert_replay_conformance_tool_attempt_marker(slow_first, "call-slow", "slow_tool");
    assert_replay_conformance_tool_attempt_marker(fast_first, "call-fast", "fast_tool");

    start_replay();
    let replay_local_calls = Arc::new(Mutex::new(Vec::new()));
    let replay_pass = if controller.supports_concurrent_effects() {
        tokio::time::timeout(std::time::Duration::from_secs(2), async {
            tokio::join!(
                controller.execute_effect(
                    fast,
                    replay_conformance_failing_executor(Arc::clone(&replay_local_calls)),
                ),
                controller.execute_effect(
                    slow,
                    replay_conformance_failing_executor(Arc::clone(&replay_local_calls)),
                ),
            )
        })
        .await
        .expect("concurrent tool-attempt replay must resolve from host history")
    } else {
        let fast_replay = controller
            .execute_effect(
                fast,
                replay_conformance_failing_executor(Arc::clone(&replay_local_calls)),
            )
            .await;
        let slow_replay = controller
            .execute_effect(
                slow,
                replay_conformance_failing_executor(Arc::clone(&replay_local_calls)),
            )
            .await;
        (fast_replay, slow_replay)
    };
    let fast_replay = replay_pass.0.expect("fast tool-attempt replay");
    let slow_replay = replay_pass.1.expect("slow tool-attempt replay");
    assert_replay_conformance_tool_attempt_marker(fast_replay, "call-fast", "fast_tool");
    assert_replay_conformance_tool_attempt_marker(slow_replay, "call-slow", "slow_tool");
    assert!(
        replay_local_calls
            .lock()
            .expect("tool-attempt replay local calls")
            .is_empty(),
        "tool-attempt replay must return recorded outcomes without invoking local executors"
    );
}

#[cfg(any(test, feature = "testing"))]
fn durable_step_conformance_envelope(
    step_id: &'static str,
    input: serde_json::Value,
) -> RuntimeEffectEnvelope {
    RuntimeEffectEnvelope::new(
        RuntimeInvocation::effect(
            RuntimeScope::for_turn("durable-step-session", "durable-step-turn", 7, 0),
            format!("durable-step:{step_id}"),
            RuntimeEffectKind::DurableStep,
            format!("durable-step-replay:{step_id}"),
        ),
        RuntimeEffectCommand::DurableStep {
            step_id: step_id.to_string(),
            input,
        },
    )
}

#[cfg(any(test, feature = "testing"))]
fn durable_step_conformance_failing_executor(
    local_calls: Arc<Mutex<Vec<String>>>,
) -> RuntimeEffectLocalExecutor<'static> {
    RuntimeEffectLocalExecutor::durable_step(move |input| async move {
        local_calls
            .lock()
            .expect("local calls")
            .push(input.to_string());
        Err(crate::RuntimeError::new(
            "durable_step_replay_local_executor_called",
            "recorded durable-step replay must not invoke local execution",
        ))
    })
}

/// One controller bound to a shared durable effect-replay store, paired with a
/// `start_replay` toggle. The toggle exists because `start_replay` is a concrete
/// controller affordance, not a [`RuntimeEffectController`] trait method.
#[cfg(any(test, feature = "testing"))]
pub struct LeaseFencingController {
    pub controller: Arc<dyn RuntimeEffectController>,
    pub start_replay: Box<dyn Fn() + Send + Sync>,
}

/// A raw mutation applied to the effect-replay row for a given `replay_key`.
/// Backends implement it with a direct row update (SQLite `rusqlite`, Postgres
/// `sqlx`); it is async so Postgres can issue a pooled query.
#[cfg(any(test, feature = "testing"))]
pub type EffectLeaseMutator = Box<
    dyn Fn(String) -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send>> + Send + Sync,
>;

/// Factory that builds a fresh lease-fencing controller bound to one shared
/// durable store with the requested lease TTL. Async so Postgres backends can
/// issue pooled queries during setup.
#[cfg(any(test, feature = "testing"))]
pub type EffectLeaseControllerFactory = Box<
    dyn Fn(
            std::time::Duration,
        )
            -> std::pin::Pin<Box<dyn std::future::Future<Output = LeaseFencingController> + Send>>
        + Send
        + Sync,
>;

/// Backend adapter for the effect-replay lease-fencing conformance suite.
///
/// `make_controller` returns a fresh controller bound to one shared durable
/// store with the requested lease TTL. `steal_lease` overwrites the lease
/// owner/token for a `replay_key` (another worker reclaimed the row);
/// `expire_lease` forces the lease already-expired. Both mutate the same store
/// the controllers share.
#[cfg(any(test, feature = "testing"))]
pub struct EffectLeaseFencingBackend {
    pub make_controller: EffectLeaseControllerFactory,
    pub steal_lease: EffectLeaseMutator,
    pub expire_lease: EffectLeaseMutator,
}

#[cfg(any(test, feature = "testing"))]
fn lease_fencing_envelope(replay_key: &str) -> RuntimeEffectEnvelope {
    RuntimeEffectEnvelope::new(
        RuntimeInvocation::effect(
            RuntimeScope::for_turn("effect-lease-session", "effect-lease-turn", 1, 0),
            replay_key,
            RuntimeEffectKind::ExecCode,
            replay_key,
        ),
        RuntimeEffectCommand::ExecCode {
            language: "code".to_string(),
            code: "emit".to_string(),
        },
    )
}

/// Run the durable effect-replay lease-fencing conformance suite. Every durable
/// effect-replay controller (SQLite, Postgres, ...) must satisfy the same
/// fencing contract, so the row-level renewal/steal/expiry behavior lives here
/// once instead of in store-specific raw-row tests:
///
/// - a renewed in-progress lease keeps a competing claimant out, then replays;
/// - a stolen lease aborts the original owner with a lease-lost error;
/// - a lease that expires before finalize is rejected with a lease-lost error;
/// - a controller constructed with a non-default short TTL actually expires on
///   that window (the configured TTL cannot be ignored).
#[cfg(any(test, feature = "testing"))]
pub async fn effect_controller_lease_fencing(backend: EffectLeaseFencingBackend) {
    let run = uuid::Uuid::new_v4().to_string();
    lease_fencing_renews_long_running_lease(&backend, &run).await;
    lease_fencing_reports_lease_lost_when_stolen(&backend, &run).await;
    lease_fencing_rejects_finalize_after_expiry(&backend, &run).await;
    lease_fencing_honors_configured_short_ttl(&backend, &run).await;
}

#[cfg(any(test, feature = "testing"))]
async fn lease_fencing_renews_long_running_lease(backend: &EffectLeaseFencingBackend, run: &str) {
    // Generous TTL: the renewal task must outlive scheduler starvation when
    // the whole workspace's test binaries run in parallel; a tight TTL makes
    // this case flake under load without exercising anything extra.
    let ttl = std::time::Duration::from_millis(300);
    let replay_key = format!("lease-renewal-{run}");
    let first = (backend.make_controller)(ttl).await;
    let second = (backend.make_controller)(ttl).await;
    let envelope = lease_fencing_envelope(&replay_key);

    let (entered_tx, entered_rx) = tokio::sync::oneshot::channel();
    let release = Arc::new(tokio::sync::Notify::new());
    let first_controller = Arc::clone(&first.controller);
    let first_envelope = envelope.clone();
    let first_release = Arc::clone(&release);
    let first_task = tokio::spawn(async move {
        first_controller
            .execute_effect(
                first_envelope,
                RuntimeEffectLocalExecutor::testing(move |_| async move {
                    let _ = entered_tx.send(());
                    first_release.notified().await;
                    Ok(replay_conformance_exec_outcome("renewed-owner"))
                }),
            )
            .await
    });
    entered_rx.await.expect("first executor entered");

    // Let several lease TTLs lapse; the renewal task must keep the lease alive.
    tokio::time::sleep(ttl * 3).await;
    let competing = tokio::time::timeout(
        ttl * 2,
        second.controller.execute_effect(
            envelope.clone(),
            RuntimeEffectLocalExecutor::testing(move |_| async move {
                Ok(replay_conformance_exec_outcome("stolen-owner"))
            }),
        ),
    )
    .await;
    assert!(
        competing.is_err(),
        "renewed in-progress lease should keep a competing claimant busy",
    );

    release.notify_waiters();
    let first_outcome = first_task
        .await
        .expect("first task joins")
        .expect("renewed owner finalizes");
    assert_replay_conformance_exec_marker(first_outcome, "renewed-owner");

    (second.start_replay)();
    let replayed = second
        .controller
        .execute_effect(
            envelope,
            replay_conformance_failing_executor(Arc::new(Mutex::new(Vec::new()))),
        )
        .await
        .expect("replayed renewed outcome");
    assert_replay_conformance_exec_marker(replayed, "renewed-owner");
}

#[cfg(any(test, feature = "testing"))]
async fn lease_fencing_reports_lease_lost_when_stolen(
    backend: &EffectLeaseFencingBackend,
    run: &str,
) {
    let ttl = std::time::Duration::from_millis(300);
    let replay_key = format!("lease-stolen-{run}");
    let controller = (backend.make_controller)(ttl).await;
    let envelope = lease_fencing_envelope(&replay_key);

    let (entered_tx, entered_rx) = tokio::sync::oneshot::channel();
    let never_release = Arc::new(tokio::sync::Notify::new());
    let owner = Arc::clone(&controller.controller);
    let owner_envelope = envelope.clone();
    let owner_release = Arc::clone(&never_release);
    let owner_task = tokio::spawn(async move {
        owner
            .execute_effect(
                owner_envelope,
                RuntimeEffectLocalExecutor::testing(move |_| async move {
                    let _ = entered_tx.send(());
                    owner_release.notified().await;
                    Ok(replay_conformance_exec_outcome("should-not-finalize"))
                }),
            )
            .await
    });
    entered_rx.await.expect("owner executor entered");

    (backend.steal_lease)(replay_key.clone()).await;

    let err = tokio::time::timeout(std::time::Duration::from_secs(2), owner_task)
        .await
        .expect("renewal should notice the stolen lease")
        .expect("owner task joins")
        .expect_err("stolen lease must fail the original owner");
    assert!(
        err.code.ends_with("_effect_replay_lease_lost"),
        "expected an effect-replay lease-lost error, got code `{}`: {}",
        err.code,
        err.message,
    );
    let _keep_notify_alive = never_release;
}

#[cfg(any(test, feature = "testing"))]
async fn lease_fencing_rejects_finalize_after_expiry(
    backend: &EffectLeaseFencingBackend,
    run: &str,
) {
    // A long TTL keeps the renewal task idle during the brief block so the
    // finalize path (not renewal) is the one that observes the expired lease.
    let ttl = std::time::Duration::from_secs(30);
    let replay_key = format!("lease-expiry-{run}");
    let controller = (backend.make_controller)(ttl).await;
    let envelope = lease_fencing_envelope(&replay_key);

    let (entered_tx, entered_rx) = tokio::sync::oneshot::channel();
    let release = Arc::new(tokio::sync::Notify::new());
    let owner = Arc::clone(&controller.controller);
    let owner_envelope = envelope.clone();
    let owner_release = Arc::clone(&release);
    let owner_task = tokio::spawn(async move {
        owner
            .execute_effect(
                owner_envelope,
                RuntimeEffectLocalExecutor::testing(move |_| async move {
                    let _ = entered_tx.send(());
                    owner_release.notified().await;
                    Ok(replay_conformance_exec_outcome("expired-owner"))
                }),
            )
            .await
    });
    entered_rx.await.expect("owner executor entered");

    (backend.expire_lease)(replay_key.clone()).await;
    release.notify_waiters();

    let err = owner_task
        .await
        .expect("owner task joins")
        .expect_err("expired lease must not finalize");
    assert!(
        err.code.ends_with("_effect_replay_lease_lost"),
        "expected an effect-replay lease-lost error, got code `{}`: {}",
        err.code,
        err.message,
    );
}

/// A controller built with a non-default short TTL must honor it: a claim
/// whose owner vanishes (no renewal, no finalize, no row mutation from the
/// test) becomes reclaimable by a peer after roughly that TTL — not after the
/// 30s default a backend might hardcode.
#[cfg(any(test, feature = "testing"))]
async fn lease_fencing_honors_configured_short_ttl(backend: &EffectLeaseFencingBackend, run: &str) {
    let ttl = std::time::Duration::from_millis(40);
    let replay_key = format!("lease-short-ttl-{run}");
    let vanished = (backend.make_controller)(ttl).await;
    let successor = (backend.make_controller)(ttl).await;
    let envelope = lease_fencing_envelope(&replay_key);

    let (entered_tx, entered_rx) = tokio::sync::oneshot::channel();
    let never_release = Arc::new(tokio::sync::Notify::new());
    let owner = Arc::clone(&vanished.controller);
    let owner_envelope = envelope.clone();
    let owner_release = Arc::clone(&never_release);
    let owner_task = tokio::spawn(async move {
        owner
            .execute_effect(
                owner_envelope,
                RuntimeEffectLocalExecutor::testing(move |_| async move {
                    let _ = entered_tx.send(());
                    owner_release.notified().await;
                    Ok(replay_conformance_exec_outcome("vanished-owner"))
                }),
            )
            .await
    });
    entered_rx.await.expect("vanished executor entered");
    // Kill the first owner mid-claim: aborting the task stops its renewal
    // loop, so the in-progress row is left to expire on the configured TTL.
    owner_task.abort();
    assert!(owner_task.await.is_err(), "vanished owner task aborts");

    let reclaimed = tokio::time::timeout(
        std::time::Duration::from_secs(5),
        successor.controller.execute_effect(
            envelope,
            RuntimeEffectLocalExecutor::testing(move |_| async move {
                Ok(replay_conformance_exec_outcome("successor-owner"))
            }),
        ),
    )
    .await
    .expect(
        "a successor must reclaim the abandoned row after the configured short TTL — \
         a backend ignoring the TTL knob would stay busy for the 30s default",
    )
    .expect("successor executes the reclaimed effect");
    assert_replay_conformance_exec_marker(reclaimed, "successor-owner");
    let _keep_notify_alive = never_release;
}

#[cfg(any(test, feature = "testing"))]
fn replay_conformance_exec_envelope(effect_id: &'static str) -> RuntimeEffectEnvelope {
    RuntimeEffectEnvelope::new(
        RuntimeInvocation::effect(
            RuntimeScope::for_turn(
                "effect-conformance-session",
                "effect-conformance-turn",
                7,
                0,
            ),
            effect_id,
            RuntimeEffectKind::ExecCode,
            format!("effect-conformance:effect-conformance-turn:{effect_id}"),
        ),
        RuntimeEffectCommand::ExecCode {
            language: "code".to_string(),
            code: format!("emit {effect_id}"),
        },
    )
}

#[cfg(any(test, feature = "testing"))]
fn replay_conformance_tool_attempt_envelope(
    effect_id: &'static str,
    call_id: &'static str,
    tool_name: &'static str,
) -> RuntimeEffectEnvelope {
    RuntimeEffectEnvelope::new(
        RuntimeInvocation::effect(
            RuntimeScope::for_turn(
                "tool-attempt-conformance-session",
                "tool-attempt-conformance-turn",
                7,
                0,
            ),
            effect_id,
            RuntimeEffectKind::ToolAttempt,
            format!("tool-attempt-conformance:tool-attempt-conformance-turn:{effect_id}"),
        ),
        RuntimeEffectCommand::ToolAttempt {
            call: crate::PreparedToolCall::from_parts(
                call_id,
                crate::ToolId::from(format!("tool:{tool_name}")),
                tool_name,
                serde_json::json!({ "call": call_id }),
                None,
                serde_json::json!({ "prepared": effect_id }),
            ),
            execution_grant: None,
            attempt: 1,
            max_attempts: 1,
        },
    )
}

#[cfg(any(test, feature = "testing"))]
fn replay_conformance_recording_executor(
    effect_id: &'static str,
    barrier: Arc<tokio::sync::Barrier>,
    release_slow: Arc<tokio::sync::Notify>,
    completion_order: Arc<Mutex<Vec<String>>>,
) -> RuntimeEffectLocalExecutor<'static> {
    RuntimeEffectLocalExecutor::testing(move |envelope| async move {
        assert_eq!(envelope.invocation.effect_id(), Some(effect_id));
        barrier.wait().await;
        if effect_id == "effect-slow" {
            release_slow.notified().await;
        } else {
            completion_order
                .lock()
                .expect("completion order")
                .push(effect_id.to_string());
            release_slow.notify_one();
        }
        if effect_id == "effect-slow" {
            completion_order
                .lock()
                .expect("completion order")
                .push(effect_id.to_string());
        }
        Ok(replay_conformance_exec_outcome(effect_id))
    })
}

#[cfg(any(test, feature = "testing"))]
type ReplayConformanceProbe = (
    Arc<tokio::sync::Barrier>,
    Arc<tokio::sync::Notify>,
    Arc<Mutex<Vec<String>>>,
);

#[cfg(any(test, feature = "testing"))]
fn replay_conformance_tool_attempt_recording_executor(
    effect_id: &'static str,
    call_id: &'static str,
    tool_name: &'static str,
    concurrent_probe: Option<ReplayConformanceProbe>,
) -> RuntimeEffectLocalExecutor<'static> {
    RuntimeEffectLocalExecutor::testing(move |envelope| async move {
        assert_eq!(envelope.invocation.effect_id(), Some(effect_id));
        if let Some((barrier, release_slow, completion_order)) = concurrent_probe {
            barrier.wait().await;
            if effect_id == "tool-attempt-slow" {
                release_slow.notified().await;
            } else {
                completion_order
                    .lock()
                    .expect("tool-attempt completion order")
                    .push(effect_id.to_string());
                release_slow.notify_one();
            }
            if effect_id == "tool-attempt-slow" {
                completion_order
                    .lock()
                    .expect("tool-attempt completion order")
                    .push(effect_id.to_string());
            }
        }
        Ok(replay_conformance_tool_attempt_outcome(call_id, tool_name))
    })
}

#[cfg(any(test, feature = "testing"))]
fn replay_conformance_failing_executor(
    replay_local_calls: Arc<Mutex<Vec<String>>>,
) -> RuntimeEffectLocalExecutor<'static> {
    RuntimeEffectLocalExecutor::testing(move |envelope| async move {
        replay_local_calls
            .lock()
            .expect("replay local calls")
            .push(envelope.invocation.effect_id().unwrap_or("").to_string());
        Err(RuntimeEffectControllerError::new(
            "conformance_replay_local_executor_called",
            "recorded replay must not invoke local effect execution",
        ))
    })
}

#[cfg(any(test, feature = "testing"))]
fn replay_conformance_tool_attempt_outcome(
    call_id: &'static str,
    tool_name: &'static str,
) -> RuntimeEffectOutcome {
    RuntimeEffectOutcome::ToolAttempt {
        launch: crate::ToolAttemptLaunch::Done {
            record: crate::ToolCallRecord {
                call_id: Some(call_id.to_string()),
                tool: tool_name.to_string(),
                args: serde_json::json!({ "call": call_id }),
                output: crate::ToolCallOutput::success(serde_json::json!({
                    "call": call_id,
                    "tool": tool_name,
                })),
                duration_ms: 0,
            },
        },
        triggers: Vec::new(),
    }
}

#[cfg(any(test, feature = "testing"))]
fn replay_conformance_exec_outcome(effect_id: &str) -> RuntimeEffectOutcome {
    RuntimeEffectOutcome::ExecCode {
        result: Ok(crate::ExecResponse {
            observations: Vec::new(),
            observation_truncation: Vec::new(),
            tool_calls: Vec::new(),
            images: Vec::new(),
            printed_images: Vec::new(),
            error: None,
            duration_ms: 0,
            terminal_finish: Some(serde_json::json!(effect_id)),
        }),
    }
}

#[cfg(any(test, feature = "testing"))]
fn assert_replay_conformance_exec_marker(outcome: RuntimeEffectOutcome, expected: &str) {
    let RuntimeEffectOutcome::ExecCode { result } = outcome else {
        panic!("expected exec-code effect outcome");
    };
    let response = result.expect("exec-code response");
    assert_eq!(
        response.terminal_finish,
        Some(serde_json::json!(expected)),
        "replayed outcome must come from the matching replay key"
    );
}

#[cfg(any(test, feature = "testing"))]
fn assert_replay_conformance_tool_attempt_marker(
    outcome: RuntimeEffectOutcome,
    expected_call_id: &str,
    expected_tool_name: &str,
) {
    let RuntimeEffectOutcome::ToolAttempt { launch, .. } = outcome else {
        panic!("expected tool-attempt effect outcome");
    };
    let crate::ToolAttemptLaunch::Done { record } = launch else {
        panic!("expected completed tool-attempt launch");
    };
    assert_eq!(record.call_id.as_deref(), Some(expected_call_id));
    assert_eq!(record.tool, expected_tool_name);
    assert_eq!(
        record.output.value_for_projection(),
        serde_json::json!({
            "call": expected_call_id,
            "tool": expected_tool_name,
        }),
        "replayed tool-attempt outcome must come from the matching replay key"
    );
}