flow: own worker flow state in runtime sessions

crates/flow/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "flow"
+version = "0.1.0"
+edition.workspace = true
+license.workspace = true
+
+[dependencies]
+decodal.workspace = true
+serde = { workspace = true, features = ["derive"] }
+serde_json.workspace = true
+sha2.workspace = true
+thiserror.workspace = true
+
+[dev-dependencies]
+pretty_assertions = "1"

crates/flow/src/builtin.rs

@@ -0,0 +1,70 @@
+use crate::{CompiledFlowDefinition, FlowCompileError, compile_flow_source};
+
+pub const CODER_REVIEW_FLOW_SLUG: &str = "coder-review";
+
+const CODER_REVIEW_FLOW_SOURCE: &str = include_str!("../../../resources/flows/coder-review.dcdl");
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct BuiltinFlowSource {
+    pub slug: &'static str,
+    /// Monotonic resource revision. Increment when built-in semantics change;
+    /// Runtime also pins the compiled content digest.
+    pub revision: u64,
+    pub path: &'static str,
+    pub content: &'static str,
+}
+
+impl BuiltinFlowSource {
+    pub fn compile(self) -> Result<CompiledFlowDefinition, FlowCompileError> {
+        compile_flow_source(self.content)
+    }
+}
+
+pub fn builtin_flow_source(slug: &str) -> Option<BuiltinFlowSource> {
+    match slug {
+        CODER_REVIEW_FLOW_SLUG => Some(BuiltinFlowSource {
+            slug: CODER_REVIEW_FLOW_SLUG,
+            revision: 1,
+            path: "builtin/flows/coder-review.dcdl",
+            content: CODER_REVIEW_FLOW_SOURCE,
+        }),
+        _ => None,
+    }
+}
+
+pub fn builtin_flow_sources() -> &'static [BuiltinFlowSource] {
+    const SOURCES: &[BuiltinFlowSource] = &[BuiltinFlowSource {
+        slug: CODER_REVIEW_FLOW_SLUG,
+        revision: 1,
+        path: "builtin/flows/coder-review.dcdl",
+        content: CODER_REVIEW_FLOW_SOURCE,
+    }];
+    SOURCES
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn every_builtin_flow_compiles_and_matches_catalog_identity() {
+        assert!(!builtin_flow_sources().is_empty());
+        for source in builtin_flow_sources() {
+            assert!(
+                source.revision > 0,
+                "built-in Flow revision must be positive"
+            );
+            let definition = source.compile().unwrap_or_else(|error| {
+                panic!(
+                    "builtin Flow {} failed to compile: {:?}",
+                    source.slug, error.diagnostics
+                )
+            });
+            assert_eq!(definition.name, source.slug);
+            assert_eq!(
+                builtin_flow_source(source.slug).map(|item| item.content),
+                Some(source.content)
+            );
+        }
+    }
+}

crates/flow/src/definition.rs

@@ -0,0 +1,709 @@
+use std::collections::{BTreeMap, BTreeSet, VecDeque};
+use std::fmt;
+use std::fmt::Write as _;
+
+use decodal::{Engine, LoadedSource, SourceLoader};
+use serde::{Deserialize, Serialize};
+use sha2::{Digest, Sha256};
+
+pub const FLOW_SCHEMA_VERSION: u32 = 1;
+pub const CANCELLED_STATE_ID: &str = "$cancelled";
+pub const CANCEL_TRANSITION_ID: &str = "$cancel";
+pub const CANCEL_CONDITION: &str = "An exceptional condition makes it impossible to continue the current instructions and reach a normal terminal state with the available tools, scope, and session context.";
+
+const MAX_SOURCE_BYTES: usize = 256 * 1024;
+const MAX_STATES: usize = 128;
+const MAX_TRANSITIONS_PER_STATE: usize = 32;
+const MAX_TEXT_BYTES: usize = 32 * 1024;
+
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
+#[serde(transparent)]
+pub struct StateId(String);
+
+impl StateId {
+    pub fn new(value: impl Into<String>) -> Result<Self, String> {
+        let value = value.into();
+        validate_identifier("state", &value)?;
+        Ok(Self(value))
+    }
+
+    pub fn as_str(&self) -> &str {
+        &self.0
+    }
+}
+
+impl fmt::Display for StateId {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter.write_str(&self.0)
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
+#[serde(transparent)]
+pub struct TransitionId(String);
+
+impl TransitionId {
+    pub fn new(value: impl Into<String>) -> Result<Self, String> {
+        let value = value.into();
+        validate_identifier("transition", &value)?;
+        Ok(Self(value))
+    }
+
+    pub fn as_str(&self) -> &str {
+        &self.0
+    }
+}
+
+impl fmt::Display for TransitionId {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter.write_str(&self.0)
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct CompiledFlowDefinition {
+    pub schema_version: u32,
+    pub name: String,
+    pub initial: StateId,
+    pub states: BTreeMap<StateId, CompiledState>,
+    pub content_digest: String,
+}
+
+impl CompiledFlowDefinition {
+    pub fn state(&self, state_id: &StateId) -> Option<&CompiledState> {
+        self.states.get(state_id)
+    }
+
+    pub fn outgoing(&self, state_id: &StateId) -> Option<&[CompiledTransition]> {
+        self.state(state_id)
+            .map(|state| state.transitions.as_slice())
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct CompiledState {
+    pub id: StateId,
+    pub instructions: String,
+    pub terminal: bool,
+    pub transitions: Vec<CompiledTransition>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct CompiledTransition {
+    pub id: TransitionId,
+    pub target: StateId,
+    pub condition: String,
+    pub synthetic: bool,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct FlowDiagnostic {
+    pub code: String,
+    pub path: String,
+    pub message: String,
+}
+
+impl FlowDiagnostic {
+    fn new(code: impl Into<String>, path: impl Into<String>, message: impl Into<String>) -> Self {
+        Self {
+            code: code.into(),
+            path: path.into(),
+            message: message.into(),
+        }
+    }
+}
+
+#[derive(Debug, thiserror::Error, Clone, PartialEq, Eq)]
+#[error("Flow definition is invalid")]
+pub struct FlowCompileError {
+    pub diagnostics: Vec<FlowDiagnostic>,
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct FlowSource {
+    schema_version: u32,
+    name: String,
+    initial: String,
+    states: BTreeMap<String, StateSource>,
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct StateSource {
+    instructions: String,
+    #[serde(default)]
+    terminal: bool,
+    #[serde(default)]
+    transitions: BTreeMap<String, TransitionSource>,
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct TransitionSource {
+    target: String,
+    condition: String,
+}
+
+pub fn compile_flow_source(content: &str) -> Result<CompiledFlowDefinition, FlowCompileError> {
+    if content.len() > MAX_SOURCE_BYTES {
+        return Err(one_diagnostic(
+            "source_too_large",
+            "$",
+            format!("Flow source exceeds {MAX_SOURCE_BYTES} bytes"),
+        ));
+    }
+
+    let mut engine = Engine::new(RejectImports);
+    let module = engine
+        .add_root_source("flow.dcdl", "flow.dcdl", content)
+        .map_err(|error| {
+            one_diagnostic("dcdl_parse", "$", format!("DCDL parsing failed: {error:?}"))
+        })?;
+    let value = engine.eval_module(module).map_err(|error| {
+        one_diagnostic(
+            "dcdl_evaluation",
+            "$",
+            format!("DCDL evaluation failed: {error:?}"),
+        )
+    })?;
+    let data = engine.materialize(&value).map_err(|error| {
+        one_diagnostic(
+            "dcdl_materialization",
+            "$",
+            format!("DCDL materialization failed: {error:?}"),
+        )
+    })?;
+    let value = decodal_data_to_json(&data);
+    let source = serde_json::from_value::<FlowSource>(value).map_err(|error| {
+        one_diagnostic(
+            "schema_decode",
+            "$",
+            format!("Flow source does not match schema: {error}"),
+        )
+    })?;
+    compile_typed_source(source, content)
+}
+
+fn compile_typed_source(
+    source: FlowSource,
+    content: &str,
+) -> Result<CompiledFlowDefinition, FlowCompileError> {
+    let mut diagnostics = Vec::new();
+    if source.schema_version != FLOW_SCHEMA_VERSION {
+        diagnostics.push(FlowDiagnostic::new(
+            "unsupported_schema_version",
+            "schema_version",
+            format!(
+                "unsupported Flow schema version {}; expected {FLOW_SCHEMA_VERSION}",
+                source.schema_version
+            ),
+        ));
+    }
+    if let Err(message) = validate_identifier("Flow", &source.name) {
+        diagnostics.push(FlowDiagnostic::new("invalid_name", "name", message));
+    }
+    if source.states.is_empty() {
+        diagnostics.push(FlowDiagnostic::new(
+            "states_empty",
+            "states",
+            "Flow must declare at least one state",
+        ));
+    }
+    if source.states.len() > MAX_STATES {
+        diagnostics.push(FlowDiagnostic::new(
+            "too_many_states",
+            "states",
+            format!("Flow declares more than {MAX_STATES} states"),
+        ));
+    }
+    if source.initial == CANCELLED_STATE_ID {
+        diagnostics.push(FlowDiagnostic::new(
+            "reserved_state",
+            "initial",
+            format!("{CANCELLED_STATE_ID} is reserved by the Flow runtime"),
+        ));
+    }
+
+    let initial = StateId::new(source.initial.clone()).unwrap_or_else(|message| {
+        diagnostics.push(FlowDiagnostic::new("invalid_initial", "initial", message));
+        StateId(source.initial.clone())
+    });
+
+    let declared_names = source.states.keys().cloned().collect::<BTreeSet<_>>();
+    if !declared_names.contains(initial.as_str()) {
+        diagnostics.push(FlowDiagnostic::new(
+            "initial_not_found",
+            "initial",
+            format!("initial state {initial:?} is not declared"),
+        ));
+    }
+    if declared_names.contains(CANCELLED_STATE_ID) {
+        diagnostics.push(FlowDiagnostic::new(
+            "reserved_state",
+            format!("states.{CANCELLED_STATE_ID}"),
+            format!("{CANCELLED_STATE_ID} is reserved by the Flow runtime"),
+        ));
+    }
+
+    let mut states = BTreeMap::new();
+    for (state_name, state_source) in source.states {
+        let state_path = format!("states.{state_name}");
+        let state_id = StateId::new(state_name.clone()).unwrap_or_else(|message| {
+            diagnostics.push(FlowDiagnostic::new(
+                "invalid_state_id",
+                state_path.clone(),
+                message,
+            ));
+            StateId(state_name.clone())
+        });
+        validate_text(
+            &mut diagnostics,
+            "instructions",
+            &format!("{state_path}.instructions"),
+            &state_source.instructions,
+            state_source.terminal,
+        );
+        if state_source.terminal && !state_source.transitions.is_empty() {
+            diagnostics.push(FlowDiagnostic::new(
+                "terminal_has_transitions",
+                format!("{state_path}.transitions"),
+                "terminal states must not declare outgoing transitions",
+            ));
+        }
+        if !state_source.terminal && state_source.transitions.is_empty() {
+            diagnostics.push(FlowDiagnostic::new(
+                "non_terminal_without_transition",
+                format!("{state_path}.transitions"),
+                "non-terminal states must declare at least one transition",
+            ));
+        }
+        if state_source.transitions.len() > MAX_TRANSITIONS_PER_STATE {
+            diagnostics.push(FlowDiagnostic::new(
+                "too_many_transitions",
+                format!("{state_path}.transitions"),
+                format!("state declares more than {MAX_TRANSITIONS_PER_STATE} transitions"),
+            ));
+        }
+
+        let mut transitions = Vec::new();
+        for (transition_name, transition_source) in state_source.transitions {
+            let transition_path = format!("{state_path}.transitions.{transition_name}");
+            if transition_name == CANCEL_TRANSITION_ID || transition_name.starts_with('$') {
+                diagnostics.push(FlowDiagnostic::new(
+                    "reserved_transition",
+                    transition_path.clone(),
+                    "transition identifiers beginning with '$' are reserved by the Flow runtime",
+                ));
+            }
+            let transition_id =
+                TransitionId::new(transition_name.clone()).unwrap_or_else(|message| {
+                    diagnostics.push(FlowDiagnostic::new(
+                        "invalid_transition_id",
+                        transition_path.clone(),
+                        message,
+                    ));
+                    TransitionId(transition_name.clone())
+                });
+            if transition_source.target == CANCELLED_STATE_ID
+                || transition_source.target.starts_with('$')
+            {
+                diagnostics.push(FlowDiagnostic::new(
+                    "reserved_transition_target",
+                    format!("{transition_path}.target"),
+                    "definition authors cannot target runtime-reserved states",
+                ));
+            }
+            let target = StateId::new(transition_source.target.clone()).unwrap_or_else(|message| {
+                diagnostics.push(FlowDiagnostic::new(
+                    "invalid_transition_target",
+                    format!("{transition_path}.target"),
+                    message,
+                ));
+                StateId(transition_source.target.clone())
+            });
+            if !declared_names.contains(target.as_str()) {
+                diagnostics.push(FlowDiagnostic::new(
+                    "transition_target_not_found",
+                    format!("{transition_path}.target"),
+                    format!("transition target {target:?} is not declared"),
+                ));
+            }
+            validate_text(
+                &mut diagnostics,
+                "condition",
+                &format!("{transition_path}.condition"),
+                &transition_source.condition,
+                false,
+            );
+            transitions.push(CompiledTransition {
+                id: transition_id,
+                target,
+                condition: transition_source.condition,
+                synthetic: false,
+            });
+        }
+        if !state_source.terminal {
+            transitions.push(CompiledTransition {
+                id: TransitionId(CANCEL_TRANSITION_ID.to_string()),
+                target: StateId(CANCELLED_STATE_ID.to_string()),
+                condition: CANCEL_CONDITION.to_string(),
+                synthetic: true,
+            });
+        }
+        states.insert(
+            state_id.clone(),
+            CompiledState {
+                id: state_id,
+                instructions: state_source.instructions,
+                terminal: state_source.terminal,
+                transitions,
+            },
+        );
+    }
+
+    if diagnostics.is_empty() {
+        validate_graph(&states, &initial, &mut diagnostics);
+    }
+    if !diagnostics.is_empty() {
+        return Err(FlowCompileError { diagnostics });
+    }
+
+    states.insert(
+        StateId(CANCELLED_STATE_ID.to_string()),
+        CompiledState {
+            id: StateId(CANCELLED_STATE_ID.to_string()),
+            instructions: String::new(),
+            terminal: true,
+            transitions: Vec::new(),
+        },
+    );
+
+    Ok(CompiledFlowDefinition {
+        schema_version: source.schema_version,
+        name: source.name,
+        initial,
+        states,
+        content_digest: content_digest(content),
+    })
+}
+
+fn validate_identifier(kind: &str, value: &str) -> Result<(), String> {
+    if value.is_empty() {
+        return Err(format!("{kind} identifier must not be empty"));
+    }
+    if value.len() > 128 {
+        return Err(format!("{kind} identifier exceeds 128 bytes"));
+    }
+    if value.starts_with('$') {
+        return Err(format!("{kind} identifier beginning with '$' is reserved"));
+    }
+    if !value
+        .chars()
+        .all(|character| character.is_ascii_alphanumeric() || matches!(character, '-' | '_'))
+    {
+        return Err(format!(
+            "{kind} identifier must contain only ASCII letters, digits, '-' or '_'"
+        ));
+    }
+    Ok(())
+}
+
+fn validate_text(
+    diagnostics: &mut Vec<FlowDiagnostic>,
+    field: &str,
+    path: &str,
+    value: &str,
+    allow_empty: bool,
+) {
+    if !allow_empty && value.trim().is_empty() {
+        diagnostics.push(FlowDiagnostic::new(
+            format!("{field}_empty"),
+            path,
+            format!("{field} must not be empty"),
+        ));
+    }
+    if value.len() > MAX_TEXT_BYTES {
+        diagnostics.push(FlowDiagnostic::new(
+            format!("{field}_too_large"),
+            path,
+            format!("{field} exceeds {MAX_TEXT_BYTES} bytes"),
+        ));
+    }
+}
+
+fn validate_graph(
+    states: &BTreeMap<StateId, CompiledState>,
+    initial: &StateId,
+    diagnostics: &mut Vec<FlowDiagnostic>,
+) {
+    if !states.contains_key(initial) {
+        return;
+    }
+    let mut reachable = BTreeSet::new();
+    let mut pending = VecDeque::from([initial.clone()]);
+    while let Some(current) = pending.pop_front() {
+        if !reachable.insert(current.clone()) {
+            continue;
+        }
+        if let Some(state) = states.get(&current) {
+            for transition in state
+                .transitions
+                .iter()
+                .filter(|transition| !transition.synthetic)
+            {
+                pending.push_back(transition.target.clone());
+            }
+        }
+    }
+    for state_id in states.keys() {
+        if !reachable.contains(state_id) {
+            diagnostics.push(FlowDiagnostic::new(
+                "unreachable_state",
+                format!("states.{state_id}"),
+                format!("state {state_id:?} is unreachable from initial state {initial:?}"),
+            ));
+        }
+    }
+
+    let terminal_states = states
+        .values()
+        .filter(|state| state.terminal)
+        .map(|state| state.id.clone())
+        .collect::<BTreeSet<_>>();
+    if terminal_states.is_empty() {
+        diagnostics.push(FlowDiagnostic::new(
+            "terminal_missing",
+            "states",
+            "Flow must declare at least one terminal state",
+        ));
+        return;
+    }
+
+    let mut reverse: BTreeMap<StateId, Vec<StateId>> = BTreeMap::new();
+    for state in states.values() {
+        for transition in state
+            .transitions
+            .iter()
+            .filter(|transition| !transition.synthetic)
+        {
+            reverse
+                .entry(transition.target.clone())
+                .or_default()
+                .push(state.id.clone());
+        }
+    }
+    let mut can_reach_terminal = terminal_states.clone();
+    let mut pending = terminal_states.into_iter().collect::<VecDeque<_>>();
+    while let Some(current) = pending.pop_front() {
+        for predecessor in reverse.get(&current).into_iter().flatten() {
+            if can_reach_terminal.insert(predecessor.clone()) {
+                pending.push_back(predecessor.clone());
+            }
+        }
+    }
+    for state_id in reachable {
+        if !can_reach_terminal.contains(&state_id) {
+            diagnostics.push(FlowDiagnostic::new(
+                "terminal_unreachable",
+                format!("states.{state_id}"),
+                format!(
+                    "state {state_id:?} cannot reach a user-declared terminal state; the graph contains a closed non-terminal path"
+                ),
+            ));
+        }
+    }
+}
+
+fn content_digest(content: &str) -> String {
+    let digest = Sha256::digest(content.as_bytes());
+    let mut encoded = String::with_capacity(7 + digest.len() * 2);
+    encoded.push_str("sha256:");
+    for byte in digest {
+        write!(&mut encoded, "{byte:02x}").expect("writing to String cannot fail");
+    }
+    encoded
+}
+
+struct RejectImports;
+
+impl SourceLoader for RejectImports {
+    fn load(
+        &mut self,
+        _current_key: Option<&str>,
+        specifier: &str,
+    ) -> decodal::Result<LoadedSource> {
+        Err(decodal::Diagnostic::new(
+            decodal::DiagnosticKind::Import,
+            decodal::Span::default(),
+            format!("Flow source imports are not enabled: {specifier}"),
+        ))
+    }
+}
+
+fn decodal_data_to_json(data: &decodal::Data) -> serde_json::Value {
+    match data {
+        decodal::Data::Bool(value) => serde_json::Value::Bool(*value),
+        decodal::Data::Int(value) => serde_json::Value::Number(serde_json::Number::from(*value)),
+        decodal::Data::Float(value) => serde_json::Number::from_f64(*value)
+            .map(serde_json::Value::Number)
+            .unwrap_or(serde_json::Value::Null),
+        decodal::Data::String(value) => serde_json::Value::String(value.clone()),
+        decodal::Data::Array(values) => {
+            serde_json::Value::Array(values.iter().map(decodal_data_to_json).collect())
+        }
+        decodal::Data::Object(fields) => serde_json::Value::Object(
+            fields
+                .iter()
+                .map(|field| (field.name.clone(), decodal_data_to_json(&field.value)))
+                .collect(),
+        ),
+    }
+}
+
+fn one_diagnostic(
+    code: impl Into<String>,
+    path: impl Into<String>,
+    message: impl Into<String>,
+) -> FlowCompileError {
+    FlowCompileError {
+        diagnostics: vec![FlowDiagnostic::new(code, path, message)],
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn valid_source() -> &'static str {
+        r#"{
+            schema_version = 1;
+            name = "coder-review";
+            initial = "code";
+            states = {
+                code = {
+                    instructions = "Implement and validate the requested change.";
+                    transitions = {
+                        review = {
+                            target = "review";
+                            condition = "Implementation and validation evidence are present.";
+                        };
+                    };
+                };
+                review = {
+                    instructions = "Read the independent review result.";
+                    transitions = {
+                        done = {
+                            target = "done";
+                            condition = "The independent reviewer approved the implementation.";
+                        };
+                        fix = {
+                            target = "fix";
+                            condition = "The independent reviewer requested concrete changes.";
+                        };
+                    };
+                };
+                fix = {
+                    instructions = "Resolve every open review finding and validate the fixes.";
+                    transitions = {
+                        review = {
+                            target = "review";
+                            condition = "The requested changes are resolved and ready for re-review.";
+                        };
+                    };
+                };
+                done = {
+                    instructions = "";
+                    terminal = true;
+                };
+            };
+        }"#
+    }
+
+    #[test]
+    fn compiles_definition_and_injects_cancel_transition() {
+        let definition = compile_flow_source(valid_source()).expect("valid Flow");
+        assert_eq!(definition.name, "coder-review");
+        assert!(definition.content_digest.starts_with("sha256:"));
+        let code = definition
+            .state(&StateId::new("code").unwrap())
+            .expect("code state");
+        assert_eq!(code.transitions.len(), 2);
+        assert!(
+            code.transitions
+                .iter()
+                .any(|transition| transition.id.as_str() == CANCEL_TRANSITION_ID
+                    && transition.synthetic)
+        );
+        let cancelled = definition
+            .state(&StateId(CANCELLED_STATE_ID.to_string()))
+            .expect("synthetic cancelled state");
+        assert!(cancelled.terminal);
+        assert!(cancelled.instructions.is_empty());
+    }
+
+    #[test]
+    fn rejects_unknown_structural_fields() {
+        let source = valid_source().replace(
+            "schema_version = 1;",
+            "schema_version = 1; unexpected = true;",
+        );
+        let error = compile_flow_source(&source).unwrap_err();
+        assert_eq!(error.diagnostics[0].code, "schema_decode");
+        assert!(error.diagnostics[0].message.contains("unexpected"));
+    }
+
+    #[test]
+    fn rejects_definition_authored_cancel_target() {
+        let source = valid_source().replace("target = \"review\";", "target = \"$cancelled\";");
+        let error = compile_flow_source(&source).unwrap_err();
+        assert!(
+            error
+                .diagnostics
+                .iter()
+                .any(|diagnostic| { diagnostic.code == "reserved_transition_target" })
+        );
+    }
+
+    #[test]
+    fn rejects_unreachable_state() {
+        let source = valid_source().replace(
+            "done = {\n                    instructions = \"\";",
+            "unused = { instructions = \"unused\"; terminal = true; };\n                done = {\n                    instructions = \"\";",
+        );
+        let error = compile_flow_source(&source).unwrap_err();
+        assert!(
+            error
+                .diagnostics
+                .iter()
+                .any(|diagnostic| diagnostic.code == "unreachable_state")
+        );
+    }
+
+    #[test]
+    fn rejects_closed_non_terminal_cycle_even_with_synthetic_cancel() {
+        let source = r#"{
+            schema_version = 1;
+            name = "closed-cycle";
+            initial = "a";
+            states = {
+                a = {
+                    instructions = "a";
+                    transitions = { to_b = { target = "b"; condition = "go b"; }; };
+                };
+                b = {
+                    instructions = "b";
+                    transitions = { to_a = { target = "a"; condition = "go a"; }; };
+                };
+                done = { instructions = ""; terminal = true; };
+            };
+        }"#;
+        let error = compile_flow_source(source).unwrap_err();
+        assert!(
+            error
+                .diagnostics
+                .iter()
+                .any(|diagnostic| diagnostic.code == "terminal_unreachable")
+        );
+    }
+}

crates/flow/src/lib.rs

@@ -0,0 +1,15 @@
+//! Declarative Flow definitions and deterministic transition coordination.
+//!
+//! DCDL is a source format only.  The Flow domain owns the typed source
+//! schema, semantic validation, immutable transition snapshots, and state
+//! transition rules.
+
+mod builtin;
+mod coordinator;
+mod definition;
+mod selector;
+
+pub use builtin::*;
+pub use coordinator::*;
+pub use definition::*;
+pub use selector::*;

crates/flow/src/selector.rs

@@ -0,0 +1,199 @@
+use std::fmt;
+use std::str::FromStr;
+
+use serde::{Deserialize, Deserializer, Serialize, Serializer};
+
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub enum FlowSelector {
+    Builtin { slug: String },
+    Workspace { slug: String },
+}
+
+impl FlowSelector {
+    pub fn builtin(slug: impl Into<String>) -> Result<Self, FlowSelectorError> {
+        let slug = slug.into();
+        validate_slug(&slug)?;
+        Ok(Self::Builtin { slug })
+    }
+
+    pub fn workspace(slug: impl Into<String>) -> Result<Self, FlowSelectorError> {
+        let slug = slug.into();
+        validate_slug(&slug)?;
+        Ok(Self::Workspace { slug })
+    }
+
+    pub fn slug(&self) -> &str {
+        match self {
+            Self::Builtin { slug } | Self::Workspace { slug } => slug,
+        }
+    }
+
+    pub fn source_kind(&self) -> FlowSourceKind {
+        match self {
+            Self::Builtin { .. } => FlowSourceKind::Builtin,
+            Self::Workspace { .. } => FlowSourceKind::Workspace,
+        }
+    }
+}
+
+impl fmt::Display for FlowSelector {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::Builtin { slug } => write!(formatter, "builtin:{slug}"),
+            Self::Workspace { slug } => write!(formatter, "workspace:{slug}"),
+        }
+    }
+}
+
+impl FromStr for FlowSelector {
+    type Err = FlowSelectorError;
+
+    fn from_str(value: &str) -> Result<Self, Self::Err> {
+        let (source, slug) = value.split_once(':').ok_or_else(|| {
+            FlowSelectorError::InvalidFormat(
+                "Flow selector must be source-qualified as builtin:<slug> or workspace:<slug>"
+                    .to_string(),
+            )
+        })?;
+        if slug.contains(':') {
+            return Err(FlowSelectorError::InvalidFormat(
+                "Flow selector must contain exactly one ':' separator".to_string(),
+            ));
+        }
+        match source {
+            "builtin" => Self::builtin(slug),
+            "workspace" => Self::workspace(slug),
+            other => Err(FlowSelectorError::UnknownSource(other.to_string())),
+        }
+    }
+}
+
+impl Serialize for FlowSelector {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        serializer.serialize_str(&self.to_string())
+    }
+}
+
+impl<'de> Deserialize<'de> for FlowSelector {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let value = String::deserialize(deserializer)?;
+        value.parse().map_err(serde::de::Error::custom)
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum FlowSourceKind {
+    Builtin,
+    Workspace,
+}
+
+impl FlowSourceKind {
+    pub fn as_str(self) -> &'static str {
+        match self {
+            Self::Builtin => "builtin",
+            Self::Workspace => "workspace",
+        }
+    }
+}
+
+impl fmt::Display for FlowSourceKind {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter.write_str(self.as_str())
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct FlowSourceResolveRequest {
+    pub selector: FlowSelector,
+}
+
+/// Immutable source snapshot resolved by Workspace authority for one Runtime.
+///
+/// This is read-only source authority. Starting or mutating a Flow instance is
+/// deliberately not part of this response; Runtime persists the snapshot in
+/// the target Worker's durable state before execution.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ResolvedFlowSource {
+    pub selector: FlowSelector,
+    pub workspace_id: String,
+    pub flow_id: String,
+    pub revision: u64,
+    pub content_digest: String,
+    pub definition: crate::CompiledFlowDefinition,
+}
+
+#[derive(Debug, thiserror::Error, Clone, PartialEq, Eq)]
+pub enum FlowSelectorError {
+    #[error("{0}")]
+    InvalidFormat(String),
+    #[error("unknown Flow selector source {0:?}")]
+    UnknownSource(String),
+    #[error("invalid Flow selector slug: {0}")]
+    InvalidSlug(String),
+}
+
+fn validate_slug(slug: &str) -> Result<(), FlowSelectorError> {
+    if slug.is_empty() {
+        return Err(FlowSelectorError::InvalidSlug(
+            "slug must not be empty".to_string(),
+        ));
+    }
+    if slug.len() > 128 {
+        return Err(FlowSelectorError::InvalidSlug(
+            "slug exceeds 128 bytes".to_string(),
+        ));
+    }
+    if !slug
+        .chars()
+        .all(|character| character.is_ascii_alphanumeric() || matches!(character, '-' | '_'))
+    {
+        return Err(FlowSelectorError::InvalidSlug(
+            "slug must contain only ASCII letters, digits, '-' or '_'".to_string(),
+        ));
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn selector_requires_explicit_known_source() {
+        assert_eq!(
+            "builtin:coder-review".parse::<FlowSelector>().unwrap(),
+            FlowSelector::Builtin {
+                slug: "coder-review".to_string()
+            }
+        );
+        assert_eq!(
+            "workspace:coder-review"
+                .parse::<FlowSelector>()
+                .unwrap()
+                .to_string(),
+            "workspace:coder-review"
+        );
+        assert!("coder-review".parse::<FlowSelector>().is_err());
+        assert!("project:coder-review".parse::<FlowSelector>().is_err());
+        assert!("builtin:bad/path".parse::<FlowSelector>().is_err());
+        assert!("builtin:a:b".parse::<FlowSelector>().is_err());
+    }
+
+    #[test]
+    fn selector_serde_is_one_canonical_string() {
+        let selector = FlowSelector::builtin("coder-review").unwrap();
+        let json = serde_json::to_string(&selector).unwrap();
+        assert_eq!(json, r#""builtin:coder-review""#);
+        assert_eq!(
+            serde_json::from_str::<FlowSelector>(&json).unwrap(),
+            selector
+        );
+    }
+}

crates/manifest/src/config.rs

@@ -85,6 +85,8 @@ pub struct FeatureConfigPartial {
     #[serde(default)]
     pub sub_worker: Option<FeatureFlagConfigPartial>,
     #[serde(default)]
+    pub flow: Option<FeatureFlagConfigPartial>,
+    #[serde(default)]
     pub worker: Option<FeatureFlagConfigPartial>,
     #[serde(default)]
     pub objective: Option<FeatureFlagConfigPartial>,
@@ -107,6 +109,7 @@ impl FeatureConfigPartial {
                 other.sub_worker,
                 FeatureFlagConfigPartial::merge,
             ),
+            flow: merge_option(self.flow, other.flow, FeatureFlagConfigPartial::merge),
             worker: merge_option(self.worker, other.worker, FeatureFlagConfigPartial::merge),
             objective: merge_option(
                 self.objective,
@@ -190,6 +193,7 @@ impl From<FeatureConfigPartial> for FeatureConfig {
                 .sub_worker
                 .map(FeatureFlagConfig::from)
                 .unwrap_or_default(),
+            flow: value.flow.map(FeatureFlagConfig::from).unwrap_or_default(),
             worker: value
                 .worker
                 .map(FeatureFlagConfig::from)
@@ -279,6 +283,7 @@ impl From<FeatureConfig> for FeatureConfigPartial {
             memory: Some(value.memory.into()),
             web: Some(value.web.into()),
             sub_worker: Some(value.sub_worker.into()),
+            flow: Some(value.flow.into()),
             worker: Some(value.worker.into()),
             objective: Some(value.objective.into()),
             manage_workdir: Some(value.manage_workdir.into()),

crates/manifest/src/lib.rs

@@ -113,6 +113,8 @@ pub struct FeatureConfig {
     #[serde(default)]
     pub sub_worker: FeatureFlagConfig,
     #[serde(default)]
+    pub flow: FeatureFlagConfig,
+    #[serde(default)]
     pub worker: FeatureFlagConfig,
     #[serde(default)]
     pub objective: FeatureFlagConfig,
@@ -131,6 +133,7 @@ impl Default for FeatureConfig {
             memory: MemoryFeatureConfig::disabled(),
             web: FeatureFlagConfig::disabled(),
             sub_worker: FeatureFlagConfig::disabled(),
+            flow: FeatureFlagConfig::disabled(),
             worker: FeatureFlagConfig::disabled(),
             objective: FeatureFlagConfig::disabled(),
             manage_workdir: FeatureFlagConfig::disabled(),

crates/manifest/src/profile.rs

@@ -912,7 +912,7 @@ fn builtin_profile_artifact(label: &str) -> Option<serde_json::Value> {
                 true,
                 true,
                 true,
-                false,
+                true,
             );
             Some(value)
         }
@@ -999,6 +999,7 @@ fn apply_role_profile(
     value["feature"]["memory"] = serde_json::json!({ "enabled": memory });
     value["feature"]["web"] = serde_json::json!({ "enabled": web });
     value["feature"]["sub_worker"] = serde_json::json!({ "enabled": sub_worker });
+    value["feature"]["flow"] = serde_json::json!({ "enabled": slug == "coder" });
     value["feature"]["worker"] =
         serde_json::json!({ "enabled": matches!(slug, "companion" | "orchestrator") });
     value["feature"]["manage_workdir"] = serde_json::json!({ "enabled": slug == "orchestrator" });
@@ -1530,7 +1531,8 @@ mod tests {
 
         let coder = resolve("coder");
         assert!(coder.feature.task.enabled);
-        assert!(!coder.feature.sub_worker.enabled);
+        assert!(coder.feature.sub_worker.enabled);
+        assert!(coder.feature.flow.enabled);
         assert!(!coder.feature.worker.enabled);
         assert!(coder.scope.allow.is_empty());
         assert!(coder.delegation_scope.allow.is_empty());
@@ -1548,6 +1550,7 @@ mod tests {
         let reviewer = resolve("reviewer");
         assert!(reviewer.feature.task.enabled);
         assert!(!reviewer.feature.sub_worker.enabled);
+        assert!(!reviewer.feature.flow.enabled);
         assert!(!reviewer.feature.worker.enabled);
         assert!(reviewer.feature.ticket.enabled);
         assert!(reviewer.feature.ticket.enabled);

crates/protocol/src/lib.rs

@@ -202,6 +202,11 @@ pub enum Segment {
     /// `[Dir: <path>]` listings; the flattened user text keeps the literal
     /// `@<path>` placeholder either way.
     FileRef { path: String },
+    /// Source-qualified request for the host to start a Flow before committing
+    /// this user input. Runtime resolves it through Workspace authority and
+    /// replaces it with the entered state's instructions before Worker-side
+    /// input resolution.
+    Flow { selector: String },
     /// Unknown variant from a newer client. Worker treats this as an
     /// unresolved input — surfaces an alert and inserts a placeholder.
     /// Round-trip is lossy: re-serializing yields `{"kind":"unknown"}`.
@@ -236,6 +241,11 @@ impl Segment {
                     out.push('@');
                     out.push_str(path);
                 }
+                Segment::Flow { selector } => {
+                    out.push_str("[Flow: ");
+                    out.push_str(selector);
+                    out.push(']');
+                }
                 Segment::Unknown => {}
             }
         }
@@ -900,6 +910,33 @@ mod tests {
         }
     }
 
+    #[test]
+    fn method_run_flow_segment_roundtrip() {
+        let method = Method::Run {
+            input: vec![
+                Segment::Flow {
+                    selector: "builtin:coder-review".to_string(),
+                },
+                Segment::text("Ticket context"),
+            ],
+        };
+        let json = serde_json::to_string(&method).unwrap();
+        assert!(json.contains(r#""kind":"flow""#));
+        assert!(json.contains(r#""selector":"builtin:coder-review""#));
+        let decoded = serde_json::from_str::<Method>(&json).unwrap();
+        assert!(matches!(
+            decoded,
+            Method::Run { input }
+                if matches!(
+                    input.as_slice(),
+                    [
+                        Segment::Flow { selector },
+                        Segment::Text { content }
+                    ] if selector == "builtin:coder-review" && content == "Ticket context"
+                )
+        ));
+    }
+
     #[test]
     fn segment_unknown_variant_decodes_as_unknown() {
         // A future client sends a segment kind this Worker has never heard of.

crates/session-store/src/lib.rs

@@ -50,7 +50,7 @@ pub use segment::{
     fork_at, restore, restore_by_segment, save_config_changed, save_delta, save_extension,
     save_run_completed, save_run_errored, save_turn_end, save_usage, save_user_input,
 };
-pub use segment_log::{LogEntry, RestoredState, SegmentOrigin, collect_state};
+pub use segment_log::{LogEntry, RestoredState, SegmentOrigin, SessionExtension, collect_state};
 pub use store::{Store, StoreError};
 pub use system_item::{SystemItem, SystemReminder, SystemReminderSource, render_worker_event};
 pub use worker_metadata::{

crates/session-store/src/segment.rs

@@ -183,6 +183,18 @@ pub fn save_user_input(
     session_id: SessionId,
     segment_id: SegmentId,
     segments: Vec<Segment>,
+) -> Result<(), StoreError> {
+    save_user_input_with_extensions(store, session_id, segment_id, segments, Vec::new())
+}
+
+/// Atomically persist one typed user submission and Runtime-owned session
+/// extensions in the same log record.
+pub fn save_user_input_with_extensions(
+    store: &impl Store,
+    session_id: SessionId,
+    segment_id: SegmentId,
+    segments: Vec<Segment>,
+    extensions: Vec<segment_log::SessionExtension>,
 ) -> Result<(), StoreError> {
     append_entry(
         store,
@@ -191,6 +203,7 @@ pub fn save_user_input(
         LogEntry::UserInput {
             ts: segment_log::now_millis(),
             segments,
+            extensions,
         },
     )
 }

crates/session-store/src/segment_log.rs

@@ -31,6 +31,21 @@ use crate::system_item::SystemItem;
 /// - `RunCompleted` / `RunErrored` — marks end of a `run()` or `resume()` call
 /// - `PausedTurnAbandoned` — explicit abandon/cancel of a paused interrupted turn
 /// - `ConfigChanged` — `RequestConfig` mutation
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+pub struct SessionExtension {
+    pub domain: String,
+    pub payload: serde_json::Value,
+}
+
+impl SessionExtension {
+    pub fn new(domain: impl Into<String>, payload: serde_json::Value) -> Self {
+        Self {
+            domain: domain.into(),
+            payload,
+        }
+    }
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(tag = "kind", rename_all = "snake_case")]
 pub enum LogEntry {
@@ -80,7 +95,15 @@ pub enum LogEntry {
     /// file refs) on segment restore.
     /// Replay flattens these into a `Item::user_message` for the worker
     /// history; the worker layer never sees segments directly.
-    UserInput { ts: u64, segments: Vec<Segment> },
+    UserInput {
+        ts: u64,
+        segments: Vec<Segment>,
+        /// Typed durable state committed atomically with this input record.
+        /// Runtime-owned Flow invocation uses this to avoid a Backend-instance
+        /// commit that can get ahead of Worker history.
+        #[serde(default, skip_serializing_if = "Vec::is_empty")]
+        extensions: Vec<SessionExtension>,
+    },
 
     /// One assistant-side item appended to history — assistant message,
     /// reasoning, or tool call. Singular: one entry per history item so
@@ -243,10 +266,19 @@ pub fn collect_state(entries: &[LogEntry]) -> RestoredState {
                 // log ends first, restore must treat the turn as interrupted.
                 state.last_run_interrupted = true;
             }
-            LogEntry::UserInput { segments, .. } => {
+            LogEntry::UserInput {
+                segments,
+                extensions,
+                ..
+            } => {
                 let text = Segment::flatten_to_text(segments);
                 state.history.push(Item::user_message(text));
                 state.user_segments.push(segments.clone());
+                state.extensions.extend(
+                    extensions
+                        .iter()
+                        .map(|extension| (extension.domain.clone(), extension.payload.clone())),
+                );
             }
             LogEntry::AssistantItem { item, .. } => {
                 state.history.push(Item::from(item.clone()));
@@ -350,6 +382,7 @@ mod tests {
             },
             LogEntry::UserInput {
                 ts: 2000,
+                extensions: vec![],
                 segments: vec![Segment::text("Hello")],
             },
             LogEntry::AssistantItem {
@@ -389,6 +422,7 @@ mod tests {
             },
             LogEntry::UserInput {
                 ts: 2001,
+                extensions: vec![],
                 segments: vec![Segment::text("run a tool")],
             },
             LogEntry::AssistantItem {
@@ -414,6 +448,7 @@ mod tests {
             },
             LogEntry::UserInput {
                 ts: 2000,
+                extensions: vec![],
                 segments: vec![Segment::text("Check weather")],
             },
             LogEntry::AssistantItem {
@@ -472,6 +507,7 @@ mod tests {
             },
             LogEntry::UserInput {
                 ts: 2000,
+                extensions: vec![],
                 segments: vec![Segment::text("hi")],
             },
             LogEntry::LlmUsage {
@@ -519,6 +555,7 @@ mod tests {
             },
             LogEntry::UserInput {
                 ts: 2000,
+                extensions: vec![],
                 segments: vec![Segment::text("hi")],
             },
         ]);
@@ -595,6 +632,7 @@ mod tests {
             },
             LogEntry::UserInput {
                 ts: 101,
+                extensions: vec![],
                 segments: vec![Segment::text("hi")],
             },
             LogEntry::TurnEnd {
@@ -705,6 +743,26 @@ mod tests {
         }
     }
 
+    #[test]
+    fn user_input_extensions_restore_with_the_same_committed_input() {
+        let segments = vec![Segment::text("Flow instructions"), Segment::text("Ticket")];
+        let entry = LogEntry::UserInput {
+            ts: 9999,
+            segments: segments.clone(),
+            extensions: vec![SessionExtension::new(
+                "flow.runtime.v1",
+                serde_json::json!({ "state": "implement", "revision": 0 }),
+            )],
+        };
+        let json = serde_json::to_string(&entry).unwrap();
+        let decoded: LogEntry = serde_json::from_str(&json).unwrap();
+        let state = collect_state(&[decoded]);
+        assert_eq!(state.user_segments, vec![segments]);
+        assert_eq!(state.extensions.len(), 1);
+        assert_eq!(state.extensions[0].0, "flow.runtime.v1");
+        assert_eq!(state.extensions[0].1["state"], "implement");
+    }
+
     /// Mixed segments survive a JSON round-trip through `LogEntry::UserInput`,
     /// and `collect_state` derives `Item::user_message` from the flattened
     /// text while preserving the original segments separately. This covers
@@ -727,6 +785,7 @@ mod tests {
         ];
         let entry = LogEntry::UserInput {
             ts: 4242,
+            extensions: vec![],
             segments: segments.clone(),
         };
         // JSON round-trip preserves the variant byte-for-byte.

crates/session-store/tests/fs_store_test.rs

@@ -36,6 +36,7 @@ fn round_trip_write_and_read() {
         },
         LogEntry::UserInput {
             ts: 2000,
+            extensions: vec![],
             segments: vec![protocol::Segment::text("Hello")],
         },
         LogEntry::AssistantItem {
@@ -214,6 +215,7 @@ fn read_entry_count_matches_append_tally() {
         },
         LogEntry::UserInput {
             ts: 2000,
+            extensions: vec![],
             segments: vec![protocol::Segment::text("Hello")],
         },
     ];
@@ -253,6 +255,7 @@ fn unterminated_utf8_tail_is_ignored_and_replaced_on_append() {
 
     let next = LogEntry::UserInput {
         ts: 2,
+        extensions: vec![],
         segments: vec![protocol::Segment::text("recovered")],
     };
     store.append(sid, segid, &next).unwrap();

crates/session-store/tests/session_test.rs

@@ -453,6 +453,7 @@ async fn session_auto_forks_on_conflict() {
     // Simulate another Worker writing to the same segment behind our back.
     let extra_entry = LogEntry::UserInput {
         ts: 9999,
+        extensions: vec![],
         segments: vec![protocol::Segment::text("Interloper")],
     };
     store.append(sid, original_segid, &extra_entry).unwrap();

crates/tools/src/lib.rs

@@ -62,6 +62,17 @@ pub fn core_builtin_tools(
     tools
 }
 
+pub fn read_only_builtin_tools(
+    session: workdir::WorkdirSessionHandle,
+) -> Vec<llm_engine::tool::ToolDefinition> {
+    debug_assert_eq!(
+        session.capabilities(),
+        workdir::WorkdirSessionCapabilities::READ_ONLY,
+        "read-only tool projection requires a read-only Workdir session"
+    );
+    core_builtin_tools(session, Tracker::new(), std::path::PathBuf::new())
+}
+
 pub fn web_builtin_tools(
     web_config: Option<manifest::WebConfig>,
 ) -> Vec<llm_engine::tool::ToolDefinition> {

crates/tui/src/input.rs

@@ -46,11 +46,23 @@ impl FileRefAtom {
     }
 }
 
+#[derive(Debug, Clone)]
+pub struct FlowRefAtom {
+    pub selector: String,
+}
+
+impl FlowRefAtom {
+    pub fn label(&self) -> String {
+        format!("[Flow: {}]", self.selector)
+    }
+}
+
 #[derive(Debug, Clone)]
 pub enum Atom {
     Char(char),
     Paste(PasteRef),
     FileRef(FileRefAtom),
+    FlowRef(FlowRefAtom),
 }
 
 impl Atom {
@@ -61,6 +73,7 @@ impl Atom {
             Atom::Char(_) => None,
             Atom::Paste(p) => Some((Style::default().fg(Color::Magenta), p.label())),
             Atom::FileRef(r) => Some((Style::default().fg(Color::Cyan), r.label())),
+            Atom::FlowRef(r) => Some((Style::default().fg(Color::Yellow), r.label())),
         }
     }
 }
@@ -89,7 +102,7 @@ enum WordKind {
 fn atom_class(atom: &Atom) -> AtomClass {
     match atom {
         Atom::Char(c) => char_class(*c),
-        Atom::Paste(_) | Atom::FileRef(_) => AtomClass::Chip,
+        Atom::Paste(_) | Atom::FileRef(_) | Atom::FlowRef(_) => AtomClass::Chip,
     }
 }
 
@@ -181,6 +194,11 @@ impl InputBuffer {
                     self.atoms
                         .push(Atom::FileRef(FileRefAtom { path: path.clone() }));
                 }
+                protocol::Segment::Flow { selector } => {
+                    self.atoms.push(Atom::FlowRef(FlowRefAtom {
+                        selector: selector.clone(),
+                    }));
+                }
                 protocol::Segment::Unknown => {
                     self.atoms
                         .extend("[unknown input segment]".chars().map(Atom::Char));
@@ -208,6 +226,7 @@ impl InputBuffer {
                 Atom::Char(c) => text.push(*c),
                 Atom::Paste(paste) => text.push_str(&paste.content),
                 Atom::FileRef(file) => text.push_str(&file.path),
+                Atom::FlowRef(flow) => text.push_str(&flow.selector),
             }
         }
         text
@@ -484,6 +503,12 @@ impl InputBuffer {
                         path: r.path.clone(),
                     });
                 }
+                Atom::FlowRef(r) => {
+                    flush_text(&mut buf, &mut out);
+                    out.push(protocol::Segment::Flow {
+                        selector: r.selector.clone(),
+                    });
+                }
             }
         }
         if !buf.is_empty() {
@@ -1194,7 +1219,7 @@ mod word_motion_tests {
         for a in &buf.atoms {
             match a {
                 Atom::Char(c) => out.push(*c),
-                Atom::Paste(_) | Atom::FileRef(_) => out.push_str("<P>"),
+                Atom::Paste(_) | Atom::FileRef(_) | Atom::FlowRef(_) => out.push_str("<P>"),
             }
         }
         out

crates/tui/src/ui.rs

@@ -1097,6 +1097,10 @@ fn chip_span_for(seg: &Segment, fallback: Style) -> (Style, String) {
             format!("[Clipboard #{id} | {chars} chars, {line_count} lines]"),
         ),
         Segment::FileRef { path } => (Style::default().fg(Color::Cyan), format!("@{path}")),
+        Segment::Flow { selector } => (
+            Style::default().fg(Color::Yellow),
+            format!("[Flow: {selector}]"),
+        ),
         Segment::Unknown => (fallback, "[unknown segment]".to_owned()),
     }
 }
@@ -1111,6 +1115,7 @@ fn segment_display_text(seg: &Segment) -> String {
             id, chars, lines, ..
         } => format!("[Clipboard #{id} | {chars} chars, {lines} lines]"),
         Segment::FileRef { path } => format!("@{path}"),
+        Segment::Flow { selector } => format!("[Flow: {selector}]"),
         Segment::Unknown => "[unknown segment]".to_owned(),
     }
 }

crates/tui/src/worker_list.rs

@@ -1233,6 +1233,7 @@ mod tests {
                 &LogEntry::UserInput {
                     ts,
                     segments: vec![protocol::Segment::text(text)],
+                    extensions: vec![],
                 },
             )
             .unwrap();

crates/workdir/src/lib.rs

@@ -11,6 +11,7 @@ mod operation;
 
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
 
 use async_trait::async_trait;
 use serde::{Deserialize, Serialize};
@@ -158,6 +159,105 @@ pub trait WorkdirSession: std::fmt::Debug + Send + Sync {
 
 pub type WorkdirSessionHandle = Arc<dyn WorkdirSession>;
 
+/// Ephemeral least-authority view over an existing Workdir session.
+///
+/// The wrapper exposes only stat/read/list/glob/grep and never forwards write,
+/// edit, command, or close authority to the underlying Worker session. Closing
+/// the wrapper is terminal for the view but deliberately leaves the owner's
+/// source session open.
+#[derive(Debug)]
+pub struct ReadOnlyWorkdirSession {
+    source: WorkdirSessionHandle,
+    closed: AtomicBool,
+}
+
+impl ReadOnlyWorkdirSession {
+    pub fn new(source: WorkdirSessionHandle) -> Self {
+        Self {
+            source,
+            closed: AtomicBool::new(false),
+        }
+    }
+
+    fn ensure_open(&self) -> Result<(), WorkdirError> {
+        if self.closed.load(Ordering::Acquire) {
+            Err(WorkdirError::Unavailable(
+                "read-only Workdir session is closed".to_string(),
+            ))
+        } else {
+            Ok(())
+        }
+    }
+}
+
+#[async_trait]
+impl WorkdirSession for ReadOnlyWorkdirSession {
+    fn workdir(&self) -> &Workdir {
+        self.source.workdir()
+    }
+
+    fn capabilities(&self) -> WorkdirSessionCapabilities {
+        WorkdirSessionCapabilities::READ_ONLY
+    }
+
+    async fn stat(&self, request: StatRequest) -> Result<StatResult, WorkdirError> {
+        self.ensure_open()?;
+        self.source.stat(request).await
+    }
+
+    async fn read(&self, request: ReadRequest) -> Result<ReadResult, WorkdirError> {
+        self.ensure_open()?;
+        self.source.read(request).await
+    }
+
+    async fn write(&self, _request: WriteRequest) -> Result<WriteResult, WorkdirError> {
+        Err(WorkdirError::Unsupported(WorkdirSessionCapability::Write))
+    }
+
+    async fn edit(&self, _request: EditRequest) -> Result<EditResult, WorkdirError> {
+        Err(WorkdirError::Unsupported(WorkdirSessionCapability::Edit))
+    }
+
+    async fn list(&self, request: ListRequest) -> Result<ListResult, WorkdirError> {
+        self.ensure_open()?;
+        self.source.list(request).await
+    }
+
+    async fn glob(&self, request: GlobRequest) -> Result<GlobResult, WorkdirError> {
+        self.ensure_open()?;
+        self.source.glob(request).await
+    }
+
+    async fn grep(&self, request: GrepRequest) -> Result<GrepResult, WorkdirError> {
+        self.ensure_open()?;
+        self.source.grep(request).await
+    }
+
+    async fn start_command(&self, _request: CommandRequest) -> Result<CommandHandle, WorkdirError> {
+        Err(WorkdirError::Unsupported(WorkdirSessionCapability::Command))
+    }
+
+    async fn command_status(&self, _handle: CommandHandle) -> Result<CommandStatus, WorkdirError> {
+        Err(WorkdirError::Unsupported(WorkdirSessionCapability::Command))
+    }
+
+    async fn command_output(
+        &self,
+        _request: CommandOutputRequest,
+    ) -> Result<CommandOutput, WorkdirError> {
+        Err(WorkdirError::Unsupported(WorkdirSessionCapability::Command))
+    }
+
+    async fn cancel_command(&self, _handle: CommandHandle) -> Result<(), WorkdirError> {
+        Err(WorkdirError::Unsupported(WorkdirSessionCapability::Command))
+    }
+
+    async fn close(&self) -> Result<(), WorkdirError> {
+        self.closed.store(true, Ordering::Release);
+        Ok(())
+    }
+}
+
 #[derive(Debug, thiserror::Error)]
 pub enum WorkdirError {
     #[error("Workdir session does not support {0:?}")]

crates/worker-runtime/Cargo.toml

@@ -28,6 +28,7 @@ base64.workspace = true
 axum = { workspace = true, optional = true }
 futures = { workspace = true, optional = true }
 decodal.workspace = true
+flow = { path = "../flow" }
 manifest.workspace = true
 protocol.workspace = true
 serde = { workspace = true, features = ["derive"] }

crates/worker-runtime/src/runtime.rs

@@ -2333,7 +2333,11 @@ fn validate_create_worker_request(request: &CreateWorkerRequest) -> Result<(), R
                 kind: format!("{:?}", input.kind),
             });
         }
-        if input.content.trim().is_empty() {
+        let has_segments = input
+            .segments
+            .as_ref()
+            .is_some_and(|segments| !segments.is_empty());
+        if input.content.trim().is_empty() && !has_segments {
             return Err(RuntimeError::InvalidRequest(
                 "initial_input.content must not be empty".to_string(),
             ));
@@ -2369,7 +2373,11 @@ fn validate_create_workspace_scope(
 }
 
 fn validate_worker_input(input: &WorkerInput) -> Result<(), RuntimeError> {
-    if !input.kind.is_empty_content_allowed() && input.content.trim().is_empty() {
+    let has_segments = input
+        .segments
+        .as_ref()
+        .is_some_and(|segments| !segments.is_empty());
+    if !input.kind.is_empty_content_allowed() && input.content.trim().is_empty() && !has_segments {
         return Err(RuntimeError::InvalidRequest(
             "worker input content must not be empty".to_string(),
         ));
@@ -2447,6 +2455,44 @@ mod tests {
     use std::sync::atomic::{AtomicU64, Ordering};
     use std::sync::{Arc, Mutex};
 
+    #[test]
+    fn typed_segments_allow_empty_flat_content() {
+        let input = WorkerInput {
+            kind: WorkerInputKind::User,
+            content: String::new(),
+            segments: Some(vec![protocol::Segment::Flow {
+                selector: "builtin:coder-review".to_string(),
+            }]),
+        };
+        assert!(validate_worker_input(&input).is_ok());
+    }
+
+    #[test]
+    fn empty_user_input_without_segments_is_rejected() {
+        let input = WorkerInput {
+            kind: WorkerInputKind::User,
+            content: String::new(),
+            segments: Some(Vec::new()),
+        };
+        assert!(matches!(
+            validate_worker_input(&input),
+            Err(RuntimeError::InvalidRequest(_))
+        ));
+    }
+
+    #[test]
+    fn typed_flow_segments_allow_empty_initial_flat_content() {
+        let mut request = task_request("flow");
+        request.initial_input = Some(WorkerInput {
+            kind: WorkerInputKind::User,
+            content: String::new(),
+            segments: Some(vec![protocol::Segment::Flow {
+                selector: "builtin:coder-review".to_string(),
+            }]),
+        });
+        assert!(validate_create_worker_request(&request).is_ok());
+    }
+
     fn task_request(_objective: &str) -> CreateWorkerRequest {
         let profile = ProfileSelector::Builtin("builtin:coder".to_string());
         let bundle = test_bundle_for_profile(profile.clone());
@@ -2533,6 +2579,7 @@ mod tests {
         restore_result: Mutex<Option<WorkerExecutionSpawnResult>>,
         restore_count: Mutex<u64>,
         contexts: Mutex<BTreeMap<WorkerId, WorkerExecutionContext>>,
+        dispatched_inputs: Mutex<Vec<WorkerInput>>,
         #[cfg(feature = "ws-server")]
         snapshots: Mutex<BTreeMap<WorkerId, protocol::Event>>,
     }
@@ -2607,8 +2654,9 @@ mod tests {
         fn dispatch_input(
             &self,
             _handle: &WorkerExecutionHandle,
-            _input: WorkerInput,
+            input: WorkerInput,
         ) -> WorkerExecutionResult {
+            self.dispatched_inputs.lock().unwrap().push(input);
             self.dispatch_result
                 .lock()
                 .unwrap()
@@ -3379,6 +3427,36 @@ mod tests {
         );
     }
 
+    #[test]
+    fn send_input_dispatches_segment_only_flow_submission() {
+        let backend = Arc::new(TestExecutionBackend::default());
+        let runtime = Runtime::with_execution_backend(
+            RuntimeOptions {
+                ..RuntimeOptions::default()
+            },
+            backend.clone(),
+        )
+        .unwrap();
+        runtime.store_config_bundle(test_bundle()).unwrap();
+        let detail = runtime.create_worker(task_request("flow segment")).unwrap();
+        let input = WorkerInput {
+            kind: WorkerInputKind::User,
+            content: String::new(),
+            segments: Some(vec![protocol::Segment::Flow {
+                selector: "builtin:coder-review".to_string(),
+            }]),
+        };
+
+        runtime
+            .send_input(&detail.worker_ref, input.clone())
+            .unwrap();
+
+        assert_eq!(
+            backend.dispatched_inputs.lock().unwrap().as_slice(),
+            &[input]
+        );
+    }
+
     #[cfg(feature = "ws-server")]
     #[test]
     fn send_input_records_protocol_observations() {

crates/worker-runtime/src/worker_backend.rs

@@ -48,7 +48,7 @@ use worker::ipc::protocol_session::{live_log_entry_event, subscribe_worker_proto
 use worker::{
     PromptLoader, RuntimeWorkspaceHttpClient, SegmentLogSink, Worker, WorkerController,
     WorkerError, WorkerFilesystemAuthority, WorkerHandle, WorkerSharedState,
-    WorkerWorkspaceContext, WorkspaceId,
+    WorkerWorkspaceContext, WorkspaceClient, WorkspaceId,
 };
 
 const DEFAULT_BACKEND_ID: &str = "worker-crate";
@@ -90,6 +90,11 @@ impl Drop for OwnedRuntimeArtifactRoot {
     }
 }
 
+pub struct RuntimeWorkerController {
+    pub handle: WorkerHandle,
+    pub workspace_client: Arc<dyn WorkspaceClient>,
+}
+
 /// Factory seam used by [`WorkerRuntimeExecutionBackend`] to construct a real
 /// controller-backed Worker for a Runtime catalog entry.
 #[async_trait]
@@ -97,12 +102,12 @@ pub trait RuntimeWorkerFactory: Send + Sync + 'static {
     async fn spawn_controller(
         &self,
         request: WorkerExecutionSpawnRequest,
-    ) -> Result<WorkerHandle, String>;
+    ) -> Result<RuntimeWorkerController, String>;
 
     async fn restore_controller(
         &self,
         request: WorkerExecutionRestoreRequest,
-    ) -> Result<WorkerHandle, String>;
+    ) -> Result<RuntimeWorkerController, String>;
 }
 
 /// Production factory that resolves a normal Worker profile and spawns it under
@@ -498,7 +503,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
     async fn spawn_controller(
         &self,
         request: WorkerExecutionSpawnRequest,
-    ) -> Result<WorkerHandle, String> {
+    ) -> Result<RuntimeWorkerController, String> {
         let worker_name = Self::runtime_worker_name(&request);
         let profile = Self::runtime_profile(&request);
         let has_local_filesystem = request.working_directory.is_some();
@@ -554,6 +559,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
                 )?
             }
         };
+        let flow_transition_enabled = manifest.feature.flow.enabled;
 
         let store_dir = self.store_dir()?;
         let session_store = FsStore::new(&store_dir).map_err(|err| {
@@ -609,23 +615,41 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
                 CompositeWorkerObservationProvider::new(providers),
             )));
         }
+        if flow_transition_enabled {
+            let report = worker
+                .install_runtime_flow_transition_feature()
+                .map_err(|error| format!("install Flow transition feature: {error}"))?;
+            if report.reports.iter().any(|report| !report.installed) {
+                return Err(format!(
+                    "install Flow transition feature failed: {:?}",
+                    report.reports
+                ));
+            }
+        }
 
+        let workspace_client = worker.workspace_client_handle();
         let runtime_base = self.runtime_base_dir()?;
         let (handle, _shutdown_rx) = WorkerController::spawn_runtime_managed(worker, &runtime_base)
             .await
             .map_err(|err| format!("failed to spawn Worker controller: {err}"))?;
+        if flow_transition_enabled {
+            handle.shared_state.enable_flow_transition();
+        }
         self.observation_hub.register(
             request.worker_ref.clone(),
             observation_workspace_id,
             &handle,
         );
-        Ok(handle)
+        Ok(RuntimeWorkerController {
+            handle,
+            workspace_client,
+        })
     }
 
     async fn restore_controller(
         &self,
         request: WorkerExecutionRestoreRequest,
-    ) -> Result<WorkerHandle, String> {
+    ) -> Result<RuntimeWorkerController, String> {
         let worker_name = Self::runtime_worker_name_for_ref(&request.worker_ref);
         let filesystem_authority = request
             .working_directory
@@ -711,6 +735,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
             }
             Err(err) => return Err(format!("failed to restore Worker from metadata: {err}")),
         };
+        let flow_transition_enabled = worker.manifest().feature.flow.enabled;
         if let Some(binding) = request.working_directory.as_ref() {
             worker.bind_workdir_session(Some(runtime_local_workdir_session(
                 &binding.working_directory.id,
@@ -740,23 +765,42 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
                 CompositeWorkerObservationProvider::new(providers),
             )));
         }
+        if flow_transition_enabled {
+            let report = worker
+                .install_runtime_flow_transition_feature()
+                .map_err(|error| format!("install Flow transition feature: {error}"))?;
+            if report.reports.iter().any(|report| !report.installed) {
+                return Err(format!(
+                    "install Flow transition feature failed: {:?}",
+                    report.reports
+                ));
+            }
+        }
 
+        let workspace_client = worker.workspace_client_handle();
         let runtime_base = self.runtime_base_dir()?;
         let (handle, _shutdown_rx) = WorkerController::spawn_runtime_managed(worker, &runtime_base)
             .await
             .map_err(|err| format!("failed to spawn restored Worker controller: {err}"))?;
+        if flow_transition_enabled {
+            handle.shared_state.enable_flow_transition();
+        }
         self.observation_hub.register(
             request.worker_ref.clone(),
             observation_workspace_id,
             &handle,
         );
-        Ok(handle)
+        Ok(RuntimeWorkerController {
+            handle,
+            workspace_client,
+        })
     }
 }
 
 struct RuntimeWorkerExecution {
     handle: WorkerHandle,
     busy: Arc<AtomicBool>,
+    workspace_client: Option<Arc<dyn WorkspaceClient>>,
 }
 
 /// `worker-runtime` execution backend backed by real `worker` crate Workers.
@@ -847,7 +891,14 @@ where
     fn get_execution(
         &self,
         handle: &WorkerExecutionHandle,
-    ) -> Result<(WorkerHandle, Arc<AtomicBool>), WorkerExecutionResult> {
+    ) -> Result<
+        (
+            WorkerHandle,
+            Arc<AtomicBool>,
+            Option<Arc<dyn WorkspaceClient>>,
+        ),
+        WorkerExecutionResult,
+    > {
         if handle.backend_id() != self.backend_id() {
             return Err(WorkerExecutionResult::rejected(
                 WorkerExecutionOperation::Input,
@@ -866,7 +917,13 @@ where
         })?;
         workers
             .get(handle.worker_ref())
-            .map(|execution| (execution.handle.clone(), execution.busy.clone()))
+            .map(|execution| {
+                (
+                    execution.handle.clone(),
+                    execution.busy.clone(),
+                    execution.workspace_client.clone(),
+                )
+            })
             .ok_or_else(|| {
                 WorkerExecutionResult::rejected(
                     WorkerExecutionOperation::Input,
@@ -899,6 +956,7 @@ where
         bridge_context: crate::execution::WorkerExecutionContext,
         handle: WorkerHandle,
         working_directory: Option<WorkingDirectoryBinding>,
+        workspace_client: Option<Arc<dyn WorkspaceClient>>,
     ) -> WorkerExecutionSpawnResult {
         let busy = Arc::new(AtomicBool::new(false));
         #[cfg(feature = "ws-server")]
@@ -956,7 +1014,14 @@ where
                 ));
             }
         };
-        workers.insert(worker_ref.clone(), RuntimeWorkerExecution { handle, busy });
+        workers.insert(
+            worker_ref.clone(),
+            RuntimeWorkerExecution {
+                handle,
+                busy,
+                workspace_client,
+            },
+        );
 
         WorkerExecutionSpawnResult::Connected {
             handle: WorkerExecutionHandle::new(worker_ref, self.backend_id()),
@@ -1166,8 +1231,8 @@ where
         let spawn_result =
             self.run_on_adapter_runtime(async move { factory.spawn_controller(request).await });
 
-        let handle = match spawn_result {
-            Ok(handle) => handle,
+        let controller = match spawn_result {
+            Ok(controller) => controller,
             Err(message) => {
                 if let (Some(materializer), Some(binding)) = (
                     self.working_directory_materializer.as_ref(),
@@ -1186,8 +1251,9 @@ where
             WorkerExecutionOperation::Spawn,
             worker_ref,
             bridge_context,
-            handle,
+            controller.handle,
             working_directory,
+            Some(controller.workspace_client),
         )
     }
 
@@ -1267,8 +1333,8 @@ where
         let restore_result =
             self.run_on_adapter_runtime(async move { factory.restore_controller(request).await });
 
-        let handle = match restore_result {
-            Ok(handle) => handle,
+        let controller = match restore_result {
+            Ok(controller) => controller,
             Err(message) => {
                 return WorkerExecutionSpawnResult::Errored(WorkerExecutionResult::errored(
                     WorkerExecutionOperation::Restore,
@@ -1281,8 +1347,9 @@ where
             WorkerExecutionOperation::Restore,
             worker_ref,
             bridge_context,
-            handle,
+            controller.handle,
             working_directory,
+            Some(controller.workspace_client),
         )
     }
 
@@ -1291,7 +1358,7 @@ where
         handle: &WorkerExecutionHandle,
         input: WorkerInput,
     ) -> WorkerExecutionResult {
-        let (worker, busy) = match self.get_execution(handle) {
+        let (worker, busy, _workspace_client) = match self.get_execution(handle) {
             Ok(execution) => execution,
             Err(mut result) => {
                 result.operation = WorkerExecutionOperation::Input;
@@ -1374,7 +1441,7 @@ where
         handle: &WorkerExecutionHandle,
         method: Method,
     ) -> WorkerExecutionResult {
-        let (worker, busy) = match self.get_execution(handle) {
+        let (worker, busy, _workspace_client) = match self.get_execution(handle) {
             Ok(execution) => execution,
             Err(mut result) => {
                 result.operation = WorkerExecutionOperation::ProtocolMethod;
@@ -1468,7 +1535,7 @@ where
     }
 
     fn cancel_worker(&self, handle: &WorkerExecutionHandle) -> WorkerExecutionResult {
-        let (worker, _busy) = match self.get_execution(handle) {
+        let (worker, _busy, _workspace_client) = match self.get_execution(handle) {
             Ok(execution) => execution,
             Err(mut result) => {
                 result.operation = WorkerExecutionOperation::Cancel;
@@ -1626,7 +1693,7 @@ mod tests {
         async fn spawn_controller(
             &self,
             request: WorkerExecutionSpawnRequest,
-        ) -> Result<WorkerHandle, String> {
+        ) -> Result<RuntimeWorkerController, String> {
             let manifest = WorkerManifest::from_toml(
                 r#"
                 [worker]
@@ -1668,7 +1735,7 @@ mod tests {
             let workspace_backend_ref =
                 RuntimeWorkspaceBackendRef::from_worker_request(&request.request);
             let workspace_context = workspace_backend_ref.worker_context(&request.worker_ref);
-            let workspace_client = workspace_context.client();
+            let workspace_client = workspace_context.client_handle();
             self.observed_workspace_clients.lock().unwrap().push((
                 workspace_client.kind().to_string(),
                 workspace_client.workspace_id().map(str::to_string),
@@ -1689,12 +1756,15 @@ mod tests {
                 WorkerController::spawn_runtime_managed(worker, &self.runtime_base)
                     .await
                     .map_err(|err| err.to_string())?;
-            Ok(handle)
+            Ok(RuntimeWorkerController {
+                handle,
+                workspace_client,
+            })
         }
         async fn restore_controller(
             &self,
             request: WorkerExecutionRestoreRequest,
-        ) -> Result<WorkerHandle, String> {
+        ) -> Result<RuntimeWorkerController, String> {
             let request = WorkerExecutionSpawnRequest {
                 worker_ref: request.worker_ref,
                 request: request.request,
@@ -1820,14 +1890,14 @@ mod tests {
         async fn spawn_controller(
             &self,
             _request: WorkerExecutionSpawnRequest,
-        ) -> Result<WorkerHandle, String> {
+        ) -> Result<RuntimeWorkerController, String> {
             Err("spawn failed".to_string())
         }
 
         async fn restore_controller(
             &self,
             _request: WorkerExecutionRestoreRequest,
-        ) -> Result<WorkerHandle, String> {
+        ) -> Result<RuntimeWorkerController, String> {
             Err("restore failed".to_string())
         }
     }
@@ -2040,6 +2110,9 @@ mod tests {
                 [engine]
                 max_tokens = 100
 
+                [feature.flow]
+                enabled = true
+
                 [[scope.allow]]
                 target = "{}"
                 permission = "write"
@@ -2060,8 +2133,13 @@ mod tests {
             )
             .unwrap();
 
-        let request = create_request("restore");
-        let handle = ProfileRuntimeWorkerFactory::new(root.path())
+        let mut request = create_request("restore");
+        request.workspace_api = Some(crate::catalog::WorkspaceApiRef {
+            workspace_id: "workspace-restore".to_string(),
+            base_url: "http://workspace.invalid".to_string(),
+            runtime_id: Some("runtime-restore".to_string()),
+        });
+        let controller = ProfileRuntimeWorkerFactory::new(root.path())
             .with_store_dir(&store_dir)
             .with_worker_metadata_dir(&worker_metadata_dir)
             .restore_controller(WorkerExecutionRestoreRequest {
@@ -2074,8 +2152,9 @@ mod tests {
             })
             .await
             .expect("pending restore should use the saved manifest snapshot");
+        assert!(controller.handle.shared_state.flow_transition_enabled());
 
-        handle.send(Method::Shutdown).await.unwrap();
+        controller.handle.send(Method::Shutdown).await.unwrap();
     }
 
     #[test]

crates/worker/Cargo.toml

@@ -31,6 +31,7 @@ minijinja = "2.19.0"
 chrono = "0.4"
 include_dir = "0.7.4"
 fs4 = { workspace = true, features = ["sync"] }
+flow = { path = "../flow" }
 libc = { workspace = true }
 schemars = { workspace = true }
 ticket = { workspace = true }

crates/worker/src/feature/builtin.rs

@@ -4,6 +4,7 @@
 //! same descriptor-approved registry path used by feature modules. They are not
 //! an external plugin-loading surface.
 
+pub mod flow_transition;
 pub mod manage_workdir;
 pub mod manage_worker;
 pub mod memory;

crates/worker/src/ipc/protocol_session.rs

@@ -84,6 +84,7 @@ mod tests {
         let segments = vec![protocol::Segment::text("hello from log")];
         let event = live_log_entry_event(LogEntry::UserInput {
             ts: session_store::segment_log::now_millis(),
+            extensions: vec![],
             segments: segments.clone(),
         })
         .expect("UserInput must be live-relevant");

crates/worker/src/prompt/catalog.rs

@@ -65,6 +65,8 @@ pub enum WorkerPrompt {
     MemoryExtractSystem,
     /// System prompt of the memory consolidation (integration + tidy) Engine.
     MemoryConsolidationSystem,
+    /// System prompt of the bounded Flow transition verifier.
+    FlowVerifierSystem,
     /// Wrapper around an incoming `Method::Notify` message injected into
     /// the next LLM request context as a transient system message.
     NotifyWrapper,
@@ -99,6 +101,7 @@ impl WorkerPrompt {
             Self::CompactSystem => "compact_system",
             Self::MemoryExtractSystem => "memory_extract_system",
             Self::MemoryConsolidationSystem => "memory_consolidation_system",
+            Self::FlowVerifierSystem => "flow_verifier_system",
             Self::NotifyWrapper => "notify_wrapper",
             Self::InterruptToolResultSummary => "interrupt_tool_result_summary",
             Self::InterruptSystemNote => "interrupt_system_note",
@@ -118,6 +121,7 @@ impl WorkerPrompt {
         WorkerPrompt::CompactSystem,
         WorkerPrompt::MemoryExtractSystem,
         WorkerPrompt::MemoryConsolidationSystem,
+        WorkerPrompt::FlowVerifierSystem,
         WorkerPrompt::NotifyWrapper,
         WorkerPrompt::InterruptToolResultSummary,
         WorkerPrompt::InterruptSystemNote,
@@ -133,6 +137,7 @@ impl WorkerPrompt {
         "compact_system",
         "memory_extract_system",
         "memory_consolidation_system",
+        "flow_verifier_system",
         "notify_wrapper",
         "interrupt_tool_result_summary",
         "interrupt_system_note",
@@ -342,6 +347,11 @@ impl PromptCatalog {
         )
     }
 
+    /// Render `WorkerPrompt::FlowVerifierSystem` (no inputs).
+    pub fn flow_verifier_system(&self) -> Result<String, CatalogError> {
+        self.render(WorkerPrompt::FlowVerifierSystem, Value::UNDEFINED)
+    }
+
     /// Render `WorkerPrompt::NotifyWrapper` with `{{ message }}`.
     pub fn notify_wrapper(&self, message: &str) -> Result<String, CatalogError> {
         self.render(WorkerPrompt::NotifyWrapper, single("message", message))

crates/worker/src/segment_log_sink.rs

@@ -252,6 +252,7 @@ mod tests {
     fn user_input(text: &str) -> LogEntry {
         LogEntry::UserInput {
             ts: now_millis(),
+            extensions: vec![],
             segments: vec![protocol::Segment::Text {
                 content: text.to_owned(),
             }],

crates/worker/src/shared_state.rs

@@ -1,3 +1,4 @@
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{OnceLock, RwLock};
 
 use protocol::WorkerStatus;
@@ -28,6 +29,7 @@ pub struct WorkerSharedState {
     /// `ListCompletions` queries without going through the controller. It is
     /// unset only in unit tests that construct `WorkerSharedState` directly.
     fs_view: OnceLock<WorkerFsView>,
+    flow_transition_enabled: AtomicBool,
 }
 
 impl WorkerSharedState {
@@ -44,6 +46,7 @@ impl WorkerSharedState {
             greeting,
             status: RwLock::new(WorkerStatus::Idle),
             fs_view: OnceLock::new(),
+            flow_transition_enabled: AtomicBool::new(false),
         }
     }
 
@@ -59,6 +62,14 @@ impl WorkerSharedState {
         self.fs_view.get()
     }
 
+    pub fn enable_flow_transition(&self) {
+        self.flow_transition_enabled.store(true, Ordering::Release);
+    }
+
+    pub fn flow_transition_enabled(&self) -> bool {
+        self.flow_transition_enabled.load(Ordering::Acquire)
+    }
+
     pub fn set_status(&self, status: WorkerStatus) {
         if let Ok(mut s) = self.status.write() {
             *s = status;

crates/worker/src/worker.rs

@@ -13,7 +13,8 @@ use llm_engine::llm_client::types::Role;
 use llm_engine::state::Mutable;
 use llm_engine::{Engine, EngineError, EngineResult, ToolOutputLimits, UsageRecord};
 use session_store::{
-    LogEntry, SegmentId, SessionId, Store, StoreError, SystemItem, segment_log, to_logged,
+    LogEntry, SegmentId, SessionExtension, SessionId, Store, StoreError, SystemItem, segment_log,
+    to_logged,
 };
 use session_store::{
     WorkerActiveSegmentRef, WorkerMetadata, WorkerMetadataStore, WorkerReclaimedChild,
@@ -80,7 +81,9 @@ use protocol::{
 use tokio::net::UnixStream;
 use tokio::sync::broadcast;
 use tokio::task::JoinHandle;
-use workdir::{LocalWorkdirSession, WorkdirSessionCapabilities, WorkdirSessionHandle};
+use workdir::{
+    LocalWorkdirSession, ReadOnlyWorkdirSession, WorkdirSessionCapabilities, WorkdirSessionHandle,
+};
 
 const RESTORE_RECONCILIATION_REACHABILITY_TIMEOUT: Duration = Duration::from_millis(500);
 
@@ -545,6 +548,7 @@ struct EmptyTurnRollbackSnapshot {
     usage_history_len: usize,
     ai_activity_count: usize,
     last_run_interrupted: bool,
+    flow_runtime_state: Option<flow::FlowRuntimeState>,
 }
 
 fn is_ai_materialized_item(item: &Item) -> bool {
@@ -624,6 +628,28 @@ where
     }
 }
 
+struct SessionFlowRuntimeStateCommitter<St: Store + Clone> {
+    writer: LogWriterHandle<St>,
+}
+
+impl<St> crate::feature::builtin::flow_transition::FlowRuntimeStateCommitter
+    for SessionFlowRuntimeStateCommitter<St>
+where
+    St: Store + Clone + Send + Sync + 'static,
+{
+    fn commit(&self, state: &flow::FlowRuntimeState) -> Result<(), String> {
+        let payload = serde_json::to_value(state)
+            .map_err(|error| format!("serialize Flow runtime state: {error}"))?;
+        self.writer
+            .append_entry(LogEntry::Extension {
+                ts: segment_log::now_millis(),
+                domain: FLOW_RUNTIME_EXTENSION_DOMAIN.to_string(),
+                payload,
+            })
+            .map_err(|error| error.to_string())
+    }
+}
+
 /// An independent agent execution unit.
 ///
 /// Holds a [`Engine`] directly and persists session state via
@@ -650,6 +676,10 @@ pub struct Worker<C: LlmClient, St: Store> {
     /// Path-free workspace identity/client context injected by Runtime/host.
     /// This never grants local filesystem authority.
     workspace_context: WorkerWorkspaceContext,
+    /// Runtime-owned durable Flow state reconstructed from Worker session
+    /// extensions on restore.
+    flow_runtime_state: Arc<Mutex<Option<flow::FlowRuntimeState>>>,
+    flow_feature_enabled: bool,
     /// Shared, atomically-swappable view of the Worker's resolved scope.
     /// Cloned into local WorkdirSession providers used by builtin tools, fs_view,
     /// and compaction so updates propagate at the next permission check.
@@ -833,6 +863,8 @@ impl<C: LlmClient + Clone + 'static, St: Store + Clone + 'static> Worker<C, St>
             filesystem_authority: self.filesystem_authority.clone(),
             workdir_session: self.workdir_session.clone(),
             workspace_context: self.workspace_context.clone(),
+            flow_runtime_state: self.flow_runtime_state.clone(),
+            flow_feature_enabled: self.flow_feature_enabled,
             scope: self.scope.clone(),
             delegation_scope: self.delegation_scope.clone(),
             hook_builder: HookRegistryBuilder::new(),
@@ -1031,6 +1063,8 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
             filesystem_authority,
             workdir_session,
             workspace_context,
+            flow_runtime_state: Arc::new(Mutex::new(None)),
+            flow_feature_enabled: false,
             scope,
             delegation_scope,
             hook_builder: HookRegistryBuilder::new(),
@@ -1317,6 +1351,81 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
         report
     }
 
+    /// Install the Flow transition feature against this Worker's Runtime-owned
+    /// durable Flow state. Workspace authority is used only when resolving a
+    /// new immutable source snapshot during Submit.
+    pub fn install_runtime_flow_transition_feature(
+        &mut self,
+    ) -> Result<FeatureRegistryInstallReport, String>
+    where
+        C: Clone + Send + Sync + 'static,
+        St: Clone + Send + Sync + 'static,
+    {
+        self.flow_feature_enabled = true;
+        let writer = LogWriterHandle {
+            store: self.store.clone(),
+            state: self.segment_state.clone(),
+            sink: self.sink.clone(),
+            in_flight: self.in_flight.clone(),
+        };
+        let coordinator = Arc::new(
+            crate::feature::builtin::flow_transition::RuntimeFlowCoordinatorClient::new(
+                self.flow_runtime_state.clone(),
+                Arc::new(SessionFlowRuntimeStateCommitter { writer }),
+            ),
+        );
+        Ok(self.install_flow_transition_feature(coordinator))
+    }
+
+    /// Install the Flow transition tool for a host-authorized active Flow instance.
+    ///
+    /// The coordinator is already bound to the authenticated Workspace/Worker;
+    /// model input cannot select an instance or Worker identity. The verifier uses
+    /// an immutable capture of the current committed segment and, for local
+    /// Workdirs, exposes only Read/Glob/Grep through a read-only session.
+    pub fn install_flow_transition_feature(
+        &mut self,
+        coordinator: Arc<dyn crate::feature::builtin::flow_transition::FlowCoordinatorClient>,
+    ) -> FeatureRegistryInstallReport
+    where
+        C: Clone + Send + Sync + 'static,
+        St: Clone + Send + Sync + 'static,
+    {
+        let location = self.segment_state.location();
+        let capture = crate::feature::builtin::flow_transition::StoreFlowParentCapture::new(
+            self.store.clone(),
+            location.session_id,
+            location.segment_id,
+        );
+        let read_only_tools = self
+            .workdir_session
+            .clone()
+            .map(|source| Arc::new(ReadOnlyWorkdirSession::new(source)) as WorkdirSessionHandle)
+            .map(tools::read_only_builtin_tools)
+            .unwrap_or_default();
+        let client = self
+            .engine
+            .as_ref()
+            .expect("worker taken during run")
+            .client()
+            .clone();
+        let verifier = Arc::new(
+            crate::feature::builtin::flow_transition::WorkerBackedFlowVerifier::new(
+                client,
+                self.manifest.clone(),
+                capture,
+                read_only_tools,
+            ),
+        );
+        let state = crate::feature::builtin::flow_transition::FlowTransitionState::new(
+            coordinator,
+            verifier,
+        );
+        self.install_features(FeatureRegistryBuilder::new().with_module(
+            crate::feature::builtin::flow_transition::FlowTransitionFeature::new(state),
+        ))
+    }
+
     /// Reference to the store.
     pub fn store(&self) -> &St {
         &self.store
@@ -1910,6 +2019,11 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
             usage_history_len,
             ai_activity_count: self.ai_activity_counter.load(Ordering::SeqCst),
             last_run_interrupted: self.engine().last_run_interrupted(),
+            flow_runtime_state: self
+                .flow_runtime_state
+                .lock()
+                .expect("flow_runtime_state poisoned")
+                .clone(),
         }
     }
 
@@ -1936,6 +2050,10 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
         self.engine_mut().truncate_history(snapshot.history_len);
         self.engine_mut()
             .set_last_run_interrupted(snapshot.last_run_interrupted);
+        *self
+            .flow_runtime_state
+            .lock()
+            .expect("flow_runtime_state poisoned") = snapshot.flow_runtime_state;
         self.user_segments.truncate(snapshot.user_segments_len);
         *self
             .pending_attachments
@@ -1956,6 +2074,102 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
         Ok(())
     }
 
+    fn prepare_flow_input(
+        &self,
+        input: Vec<Segment>,
+    ) -> Result<(Vec<Segment>, Option<flow::FlowRuntimeState>), WorkerError> {
+        let flow_segments = input
+            .iter()
+            .filter_map(|segment| match segment {
+                Segment::Flow { selector } => Some(selector.as_str()),
+                _ => None,
+            })
+            .collect::<Vec<_>>();
+        if flow_segments.is_empty() {
+            return Ok((input, None));
+        }
+        if flow_segments.len() != 1 {
+            return Err(WorkerError::FlowInput(
+                "one Submit may contain at most one Flow segment".to_string(),
+            ));
+        }
+        if !self.flow_feature_enabled {
+            return Err(WorkerError::FlowInput(
+                "resolved Profile does not enable feature.flow".to_string(),
+            ));
+        }
+        if self
+            .flow_runtime_state
+            .lock()
+            .expect("flow_runtime_state poisoned")
+            .as_ref()
+            .is_some_and(|state| state.instance.status == flow::FlowInstanceStatus::Active)
+        {
+            return Err(WorkerError::FlowInput(
+                "Worker already has an active Flow instance".to_string(),
+            ));
+        }
+
+        let selector = flow_segments[0]
+            .parse::<flow::FlowSelector>()
+            .map_err(|error| WorkerError::FlowInput(error.to_string()))?;
+        let workspace = self.workspace_context.client_handle();
+        if !workspace.is_available() {
+            return Err(WorkerError::FlowInput(
+                "Workspace client is unavailable".to_string(),
+            ));
+        }
+        let workspace_id = workspace
+            .workspace_id()
+            .filter(|workspace_id| !workspace_id.trim().is_empty())
+            .ok_or_else(|| {
+                WorkerError::FlowInput("Workspace client has no Workspace scope".to_string())
+            })?;
+        let request = flow::FlowSourceResolveRequest {
+            selector: selector.clone(),
+        };
+        let response = workspace
+            .execute(WorkspaceRequest {
+                method: WorkspaceRequestMethod::Post,
+                path: format!("/api/w/{workspace_id}/flows/resolve"),
+                body: Some(serde_json::to_string(&request).map_err(|error| {
+                    WorkerError::FlowInput(format!("serialize Flow source request: {error}"))
+                })?),
+            })
+            .map_err(|error| WorkerError::FlowInput(error.to_string()))?;
+        if !(200..300).contains(&response.status) {
+            let message = serde_json::from_str::<serde_json::Value>(&response.body)
+                .ok()
+                .and_then(|body| {
+                    body.get("error")
+                        .and_then(serde_json::Value::as_str)
+                        .map(ToOwned::to_owned)
+                })
+                .unwrap_or_else(|| "Workspace rejected Flow source resolution".to_string());
+            return Err(WorkerError::FlowInput(message));
+        }
+        let source: flow::ResolvedFlowSource = serde_json::from_str(&response.body)
+            .map_err(|error| WorkerError::FlowInput(format!("decode Flow source: {error}")))?;
+        if source.workspace_id != workspace_id || source.selector != selector {
+            return Err(WorkerError::FlowInput(
+                "Workspace returned a Flow source outside the requested scope".to_string(),
+            ));
+        }
+        let (state, initial_instructions) =
+            flow::FlowRuntimeState::start(&source, uuid::Uuid::now_v7().to_string())
+                .map_err(|error| WorkerError::FlowInput(error.to_string()))?;
+        let input = input
+            .into_iter()
+            .map(|segment| match segment {
+                Segment::Flow { .. } => Segment::Text {
+                    content: initial_instructions.clone(),
+                },
+                other => other,
+            })
+            .collect();
+        Ok((input, Some(state)))
+    }
+
     /// Send user input and run until the LLM turn completes.
     ///
     /// `input` is a typed segment list (see [`protocol::Segment`]). The
@@ -1968,6 +2182,20 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
     /// the Engine is aborted, history is compacted, and execution resumes
     /// automatically.
     pub async fn run(&mut self, input: Vec<Segment>) -> Result<WorkerRunResult, WorkerError> {
+        let (input, pending_flow_state) = self.prepare_flow_input(input)?;
+        let input_extensions = pending_flow_state
+            .as_ref()
+            .map(|state| {
+                serde_json::to_value(state)
+                    .map(|payload| SessionExtension::new(FLOW_RUNTIME_EXTENSION_DOMAIN, payload))
+                    .map_err(|error| {
+                        WorkerError::FlowInput(format!("serialize Flow runtime state: {error}"))
+                    })
+            })
+            .transpose()?
+            .into_iter()
+            .collect();
+
         // Paused→Run transition: if the previous turn was cut short,
         // any `Item::ToolCall` whose tool never produced a matching
         // `ToolResult` is closed with a synthetic one, and a short
@@ -1998,7 +2226,14 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
         self.commit_entry(LogEntry::UserInput {
             ts: segment_log::now_millis(),
             segments: input.clone(),
+            extensions: input_extensions,
         })?;
+        if let Some(state) = pending_flow_state {
+            *self
+                .flow_runtime_state
+                .lock()
+                .expect("flow_runtime_state poisoned") = Some(state);
+        }
         self.user_segments.push(input.clone());
 
         // Resolve `@<path>` file refs to system messages stashed for the
@@ -2150,6 +2385,15 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
         for seg in segments {
             match seg {
                 Segment::Text { .. } | Segment::Paste { .. } | Segment::FileRef { .. } => {}
+                Segment::Flow { selector } => {
+                    self.alert(
+                        AlertLevel::Error,
+                        AlertSource::Worker,
+                        format!(
+                            "received unresolved Flow invocation {selector:?}; Runtime must resolve Flow segments through Workspace authority before Worker input"
+                        ),
+                    );
+                }
                 Segment::Unknown => {
                     self.alert(
                         AlertLevel::Warn,
@@ -3043,7 +3287,23 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
                 at_turn_index: source_turn_count,
             }),
         };
-        let initial_entries = vec![entry.clone()];
+        let mut initial_entries = vec![entry.clone()];
+        if let Some(flow_state) = self
+            .flow_runtime_state
+            .lock()
+            .expect("flow_runtime_state poisoned")
+            .as_ref()
+        {
+            initial_entries.push(LogEntry::Extension {
+                ts: segment_log::now_millis(),
+                domain: FLOW_RUNTIME_EXTENSION_DOMAIN.to_string(),
+                payload: serde_json::to_value(flow_state).map_err(|error| {
+                    WorkerError::InvalidState(format!(
+                        "serialize Flow runtime state during compaction: {error}"
+                    ))
+                })?,
+            });
+        }
         self.store
             .create_segment(old_loc.session_id, new_segment_id, &initial_entries)?;
         self.segment_state.set_location(SegmentLocation {
@@ -3052,12 +3312,10 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
         });
         self.segment_state
             .set_entries_written(initial_entries.len());
-        let session_start = entry;
-        // Broadcast the SegmentStart through the sink. This atomically
-        // resets the mirror to the replacement segment prefix so any subscriber
-        // querying after this point sees the post-compaction prefix, including
-        // durable extension state.
-        self.sink.reset_with_initial_entries(vec![session_start]);
+        // Broadcast the complete compacted prefix. Runtime-owned extensions
+        // must remain visible and restorable with the replacement segment.
+        self.sink
+            .reset_with_initial_entries(initial_entries.clone());
         // Keep workers.json pointing at the live segment_id. Without this
         // a concurrent `restore_from_manifest(new_segment_id)` would
         // see no live writer and grab the session this Worker just moved
@@ -3941,6 +4199,8 @@ where
             filesystem_authority: common.filesystem_authority,
             workdir_session,
             workspace_context: common.workspace_context,
+            flow_runtime_state: Arc::new(Mutex::new(None)),
+            flow_feature_enabled: false,
             scope,
             delegation_scope: common.delegation_scope,
             hook_builder: HookRegistryBuilder::new(),
@@ -4016,6 +4276,8 @@ where
             filesystem_authority: common.filesystem_authority,
             workdir_session,
             workspace_context: common.workspace_context,
+            flow_runtime_state: Arc::new(Mutex::new(None)),
+            flow_feature_enabled: false,
             scope,
             delegation_scope: common.delegation_scope,
             hook_builder: HookRegistryBuilder::new(),
@@ -4125,6 +4387,8 @@ where
             filesystem_authority: common.filesystem_authority,
             workdir_session,
             workspace_context: common.workspace_context,
+            flow_runtime_state: Arc::new(Mutex::new(None)),
+            flow_feature_enabled: false,
             scope,
             delegation_scope: common.delegation_scope,
             hook_builder: HookRegistryBuilder::new(),
@@ -4418,6 +4682,10 @@ where
             filesystem_authority: common.filesystem_authority,
             workdir_session,
             workspace_context: common.workspace_context,
+            flow_runtime_state: Arc::new(Mutex::new(restored_flow_runtime_state(
+                &state.extensions,
+            )?)),
+            flow_feature_enabled: false,
             scope,
             delegation_scope: common.delegation_scope,
             hook_builder: HookRegistryBuilder::new(),
@@ -4896,7 +5164,7 @@ fn build_rewind_targets(segment_id: uuid::Uuid, entries: &[LogEntry]) -> Vec<Rew
     let mut turn_index = 0usize;
     let mut targets = Vec::new();
     for (entry_index, entry) in entries.iter().enumerate() {
-        if let LogEntry::UserInput { segments, ts } = entry {
+        if let LogEntry::UserInput { segments, ts, .. } = entry {
             turn_index += 1;
             let truncate_entries = rewind_truncate_entries(entries, entry_index);
             let tool_warning = suffix_has_tool_side_effects(&entries[truncate_entries..]);
@@ -4960,6 +5228,11 @@ fn preview_segments(segments: &[Segment]) -> String {
                 preview.push('@');
                 preview.push_str(path);
             }
+            Segment::Flow { selector } => {
+                preview.push_str("[Flow: ");
+                preview.push_str(selector);
+                preview.push(']');
+            }
             Segment::Unknown => preview.push_str("[unknown input segment]"),
         }
     }
@@ -4972,8 +5245,31 @@ fn preview_segments(segments: &[Segment]) -> String {
     out
 }
 
+const FLOW_RUNTIME_EXTENSION_DOMAIN: &str = "flow.runtime.v1";
+
+fn restored_flow_runtime_state(
+    extensions: &[(String, serde_json::Value)],
+) -> Result<Option<flow::FlowRuntimeState>, WorkerError> {
+    extensions
+        .iter()
+        .rev()
+        .find(|(domain, _)| domain == FLOW_RUNTIME_EXTENSION_DOMAIN)
+        .map(|(_, payload)| {
+            serde_json::from_value(payload.clone()).map_err(|error| {
+                WorkerError::InvalidState(format!("invalid persisted Flow runtime state: {error}"))
+            })
+        })
+        .transpose()
+}
+
 #[derive(Debug, thiserror::Error)]
 pub enum WorkerError {
+    #[error("invalid durable Worker state: {0}")]
+    InvalidState(String),
+
+    #[error("Flow input rejected: {0}")]
+    FlowInput(String),
+
     #[error(transparent)]
     Engine(#[from] EngineError),
 
@@ -5892,6 +6188,64 @@ mod build_summary_prompt_tests {
         }
     }
 
+    #[derive(Debug, Default)]
+    struct FlowSourceWorkspaceClient {
+        requests: Mutex<Vec<WorkspaceRequest>>,
+    }
+
+    impl WorkspaceClient for FlowSourceWorkspaceClient {
+        fn workspace_id(&self) -> Option<&str> {
+            Some("workspace-test")
+        }
+
+        fn kind(&self) -> &str {
+            "flow-source-test"
+        }
+
+        fn is_available(&self) -> bool {
+            true
+        }
+
+        fn execute(
+            &self,
+            request: WorkspaceRequest,
+        ) -> Result<WorkspaceResponse, WorkspaceClientError> {
+            self.requests
+                .lock()
+                .expect("Flow source request lock")
+                .push(request);
+            let definition = flow::compile_flow_source(
+                r#"{
+                    schema_version = 1;
+                    name = "coder-review";
+                    initial = "implement";
+                    states = {
+                        implement = {
+                            instructions = "Implement the Ticket and request review.";
+                            transitions = {
+                                done = { target = "done"; condition = "The work is approved."; };
+                            };
+                        };
+                        done = { instructions = "Complete."; terminal = true; };
+                    };
+                }"#,
+            )
+            .unwrap();
+            let source = flow::ResolvedFlowSource {
+                selector: "builtin:coder-review".parse().unwrap(),
+                workspace_id: "workspace-test".to_string(),
+                flow_id: "flow-source-1".to_string(),
+                revision: 3,
+                content_digest: definition.content_digest.clone(),
+                definition,
+            };
+            Ok(WorkspaceResponse {
+                status: 200,
+                body: serde_json::to_string(&source).unwrap(),
+            })
+        }
+    }
+
     #[derive(Clone)]
     struct NoopClient;
 
@@ -5927,6 +6281,154 @@ mod build_summary_prompt_tests {
         }
     }
 
+    #[tokio::test]
+    async fn flow_transition_feature_installs_runtime_local_coordinator() {
+        let dir = tempfile::tempdir().unwrap();
+        let store = session_store::FsStore::new(dir.path().join("sessions")).unwrap();
+        let workspace_client = Arc::new(RecordingAuditWorkspaceClient::default());
+        let mut worker = Worker::new(
+            minimal_manifest(),
+            Engine::new(NoopClient),
+            store,
+            WorkerWorkspaceContext::with_client(
+                Some(WorkspaceId::new("workspace-test").unwrap()),
+                workspace_client,
+            ),
+            WorkerFilesystemAuthority::None,
+            Scope::empty(),
+        )
+        .await
+        .unwrap();
+        worker.ensure_segment_head().unwrap();
+        let report = worker
+            .install_runtime_flow_transition_feature()
+            .expect("scoped Workspace Flow feature");
+        assert_eq!(report.reports.len(), 1);
+        assert!(report.reports[0].installed);
+        assert_eq!(
+            report.reports[0]
+                .installed_tools
+                .iter()
+                .map(String::as_str)
+                .collect::<Vec<_>>(),
+            ["RequestFlowTransition"]
+        );
+    }
+
+    #[tokio::test]
+    async fn flow_submit_persists_runtime_state_atomically_with_worker_input() {
+        let dir = tempfile::tempdir().unwrap();
+        let store = session_store::FsStore::new(dir.path().join("sessions")).unwrap();
+        let workspace_client = Arc::new(FlowSourceWorkspaceClient::default());
+        let mut worker = Worker::new(
+            minimal_manifest(),
+            Engine::new(NoopClient),
+            store.clone(),
+            WorkerWorkspaceContext::with_client(
+                Some(WorkspaceId::new("workspace-test").unwrap()),
+                workspace_client.clone(),
+            ),
+            WorkerFilesystemAuthority::None,
+            Scope::empty(),
+        )
+        .await
+        .unwrap();
+        worker.ensure_segment_head().unwrap();
+        let disabled = worker.prepare_flow_input(vec![Segment::Flow {
+            selector: "builtin:coder-review".to_string(),
+        }]);
+        assert!(
+            matches!(disabled, Err(WorkerError::FlowInput(message)) if message.contains("feature.flow"))
+        );
+        worker.install_runtime_flow_transition_feature().unwrap();
+        let multiple = worker.prepare_flow_input(vec![
+            Segment::Flow {
+                selector: "builtin:coder-review".to_string(),
+            },
+            Segment::Flow {
+                selector: "workspace:coder-review".to_string(),
+            },
+        ]);
+        assert!(
+            matches!(multiple, Err(WorkerError::FlowInput(message)) if message.contains("at most one"))
+        );
+        let invalid = worker.prepare_flow_input(vec![Segment::Flow {
+            selector: "coder-review".to_string(),
+        }]);
+        assert!(matches!(invalid, Err(WorkerError::FlowInput(_))));
+
+        let (segments, state) = worker
+            .prepare_flow_input(vec![
+                Segment::Flow {
+                    selector: "builtin:coder-review".to_string(),
+                },
+                Segment::text("Implement Ticket 00001"),
+            ])
+            .unwrap();
+        let state = state.expect("new Flow runtime state");
+        let extension = SessionExtension::new(
+            FLOW_RUNTIME_EXTENSION_DOMAIN,
+            serde_json::to_value(&state).unwrap(),
+        );
+        worker
+            .commit_entry(LogEntry::UserInput {
+                ts: segment_log::now_millis(),
+                segments: segments.clone(),
+                extensions: vec![extension],
+            })
+            .unwrap();
+        *worker
+            .flow_runtime_state
+            .lock()
+            .expect("flow runtime state lock") = Some(state.clone());
+
+        assert!(matches!(segments[0], Segment::Text { .. }));
+        assert_eq!(segments[1], Segment::text("Implement Ticket 00001"));
+        assert_eq!(state.instance.definition_revision, 3);
+        assert_eq!(state.instance.current_state.as_str(), "implement");
+        assert_eq!(workspace_client.requests.lock().unwrap().len(), 1);
+
+        let location = worker.segment_state.location();
+        let restored = session_store::collect_state(
+            &store
+                .read_all(location.session_id, location.segment_id)
+                .unwrap(),
+        );
+        let restored_flow = restored_flow_runtime_state(&restored.extensions)
+            .unwrap()
+            .expect("restored Flow state");
+        assert_eq!(restored_flow, state);
+        assert_eq!(restored.user_segments, vec![segments]);
+
+        let duplicate = worker.prepare_flow_input(vec![Segment::Flow {
+            selector: "builtin:coder-review".to_string(),
+        }]);
+        assert!(
+            matches!(duplicate, Err(WorkerError::FlowInput(message)) if message.contains("active Flow"))
+        );
+
+        let mut detached = Worker::new(
+            minimal_manifest(),
+            Engine::new(NoopClient),
+            session_store::FsStore::new(dir.path().join("detached-sessions")).unwrap(),
+            WorkerWorkspaceContext::unavailable(
+                Some(WorkspaceId::new("workspace-test").unwrap()),
+                "test unavailable",
+            ),
+            WorkerFilesystemAuthority::None,
+            Scope::empty(),
+        )
+        .await
+        .unwrap();
+        detached.install_runtime_flow_transition_feature().unwrap();
+        let unavailable = detached.prepare_flow_input(vec![Segment::Flow {
+            selector: "builtin:coder-review".to_string(),
+        }]);
+        assert!(
+            matches!(unavailable, Err(WorkerError::FlowInput(message)) if message.contains("unavailable"))
+        );
+    }
+
     async fn rewind_test_worker() -> (
         tempfile::TempDir,
         Worker<NoopClient, session_store::FsStore>,
@@ -5972,6 +6474,7 @@ mod build_summary_prompt_tests {
             worker,
             LogEntry::UserInput {
                 ts: ts + 1,
+                extensions: vec![],
                 segments: vec![text_segment(text)],
             },
         );
@@ -6603,6 +7106,7 @@ mod build_summary_prompt_tests {
         worker
             .commit_entry(LogEntry::UserInput {
                 ts: segment_log::now_millis(),
+                extensions: vec![],
                 segments: vec![text_segment(
                     "The cancellation regression must leave this evidence available for retry.",
                 )],

crates/worker/tests/compact_events_test.rs

@@ -310,6 +310,7 @@ permission = "write"
             &LogEntry::UserInput {
                 ts: 9999,
                 segments: vec![protocol::Segment::text("interloper")],
+                extensions: vec![],
             },
         )
         .unwrap();

crates/workspace-server/Cargo.toml

@@ -19,6 +19,7 @@ async-trait.workspace = true
 axum = { workspace = true, features = ["ws"] }
 chrono = { version = "0.4", default-features = false, features = ["clock"] }
 futures.workspace = true
+flow = { path = "../flow" }
 manifest.workspace = true
 protocol = { workspace = true }
 project-record.workspace = true

crates/workspace-server/src/hosts.rs

@@ -221,6 +221,8 @@ pub struct HostSummary {
 pub struct WorkerWorkspaceSummary {
     pub visibility: String,
     pub identity: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub workspace_id: Option<String>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
@@ -1495,6 +1497,7 @@ impl EmbeddedWorkerRuntime {
             workspace: WorkerWorkspaceSummary {
                 visibility: "backend_internal".to_string(),
                 identity: "runtime_registry_worker".to_string(),
+                workspace_id: summary.workspace_id.clone(),
             },
             state: embedded_worker_status_label(summary.status).to_string(),
             last_seen_at: None,
@@ -1533,6 +1536,7 @@ impl EmbeddedWorkerRuntime {
             workspace: WorkerWorkspaceSummary {
                 visibility: "backend_internal".to_string(),
                 identity: "runtime_registry_worker".to_string(),
+                workspace_id: detail.workspace_id.clone(),
             },
             state: embedded_worker_status_label(detail.status).to_string(),
             last_seen_at: None,
@@ -2575,6 +2579,7 @@ impl RemoteWorkerRuntime {
             workspace: WorkerWorkspaceSummary {
                 visibility: "remote_runtime".to_string(),
                 identity: "runtime_registry_worker".to_string(),
+                workspace_id: summary.workspace_id.clone(),
             },
             state: embedded_worker_status_label(summary.status).to_string(),
             last_seen_at: None,
@@ -2617,6 +2622,7 @@ impl RemoteWorkerRuntime {
             workspace: WorkerWorkspaceSummary {
                 visibility: "remote_runtime".to_string(),
                 identity: "runtime_registry_worker".to_string(),
+                workspace_id: detail.workspace_id.clone(),
             },
             state: embedded_worker_status_label(detail.status).to_string(),
             last_seen_at: None,
@@ -3885,6 +3891,7 @@ pub fn placeholder_worker(host_id: impl Into<String>) -> WorkerSummary {
         workspace: WorkerWorkspaceSummary {
             visibility: "none".to_string(),
             identity: "unsupported".to_string(),
+            workspace_id: None,
         },
         state: "unsupported".to_string(),
         last_seen_at: None,
@@ -4329,6 +4336,7 @@ mod tests {
                     workspace: WorkerWorkspaceSummary {
                         visibility: "opaque".to_string(),
                         identity: host_id.to_string(),
+                        workspace_id: None,
                     },
                     state: "available".to_string(),
                     last_seen_at: None,

crates/workspace-server/src/lib.rs

@@ -57,6 +57,8 @@ pub enum Error {
     Ticket(#[from] ticket::TicketError),
     #[error("yaml error: {0}")]
     Yaml(#[from] serde_yaml::Error),
+    #[error("invalid input: {0}")]
+    InvalidInput(String),
     #[error("invalid project record id `{0}`")]
     InvalidRecordId(String),
     #[error("workspace backend config error: {0}")]

crates/workspace-server/src/server.rs

@@ -12,6 +12,7 @@ use axum::response::{IntoResponse, Response};
 use axum::routing::{delete, get, patch, post, put};
 use axum::{Json, Router};
 use chrono::{Duration, SecondsFormat, Utc};
+use flow::{FlowSourceKind, FlowSourceResolveRequest, ResolvedFlowSource};
 use futures::{SinkExt, StreamExt};
 use memory::backend::{
     MemoryBackendHttpResponse, MemoryBackendOperation, MemoryConsolidateStagingOperation,
@@ -96,9 +97,9 @@ use crate::runtime_subscription::RuntimeSubscriptionBroker;
 use crate::skills;
 use crate::store::{
     AccountRecord, ApiTokenRecord, AuthChallengeRecord, BrowserSessionRecord, ControlPlaneStore,
-    DeviceLoginFlowRecord, PasskeyCredentialRecord, RepositoryRecord, TicketWorkerAssignmentRecord,
-    UserRecord, WorkdirRegistryRecord, WorkerRegistryRecord, WorkerWorkdirLinkRecord,
-    WorkspaceRecord,
+    DeviceLoginFlowRecord, FlowSourceRecord, PasskeyCredentialRecord, RepositoryRecord,
+    TicketWorkerAssignmentRecord, UserRecord, WorkdirRegistryRecord, WorkerRegistryRecord,
+    WorkerWorkdirLinkRecord, WorkspaceRecord,
 };
 use crate::{Error, Result};
 use worker_runtime::catalog::{
@@ -700,6 +701,18 @@ pub fn build_router(api: WorkspaceApi) -> Router {
                 .put(scoped_update_profile_source)
                 .delete(scoped_delete_profile_source),
         )
+        .route(
+            "/api/w/{workspace_id}/flows",
+            get(scoped_list_flows).put(scoped_put_flow),
+        )
+        .route(
+            "/api/w/{workspace_id}/flows/resolve",
+            post(scoped_resolve_flow_source),
+        )
+        .route(
+            "/api/w/{workspace_id}/flows/{flow_id}",
+            get(scoped_get_flow),
+        )
         .route("/api/tickets", get(list_tickets))
         .route(
             "/api/w/{workspace_id}/tickets",
@@ -1669,6 +1682,19 @@ struct ScopedWorkspacePath {
     workspace_id: String,
 }
 
+#[derive(Debug, Deserialize)]
+struct ScopedFlowPath {
+    workspace_id: String,
+    flow_id: String,
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct PutFlowRequest {
+    path: String,
+    content: String,
+}
+
 #[derive(Debug, Deserialize)]
 #[serde(deny_unknown_fields)]
 struct AttachCurrentWorkerWorkdirRequest {
@@ -1753,6 +1779,107 @@ fn validate_workspace_scope(api: &WorkspaceApi, workspace_id: &str) -> ApiResult
     }
 }
 
+async fn scoped_list_flows(
+    State(api): State<WorkspaceApi>,
+    AxumPath(path): AxumPath<ScopedWorkspacePath>,
+) -> ApiResult<Json<Vec<FlowSourceRecord>>> {
+    validate_workspace_scope(&api, &path.workspace_id)?;
+    Ok(Json(api.store.list_flow_sources(&path.workspace_id)?))
+}
+
+async fn scoped_put_flow(
+    State(api): State<WorkspaceApi>,
+    AxumPath(path): AxumPath<ScopedWorkspacePath>,
+    Json(request): Json<PutFlowRequest>,
+) -> ApiResult<Json<FlowSourceRecord>> {
+    validate_workspace_scope(&api, &path.workspace_id)?;
+    let definition = flow::compile_flow_source(&request.content).map_err(|error| {
+        Error::InvalidInput(format!(
+            "invalid Flow source: {}",
+            serde_json::to_string(&error.diagnostics)
+                .unwrap_or_else(|_| "diagnostics unavailable".to_string())
+        ))
+    })?;
+    let expected_path = format!("flows/{}.dcdl", definition.name);
+    if request.path != expected_path {
+        return Err(
+            Error::InvalidInput(format!("Flow source path must be `{expected_path}`")).into(),
+        );
+    }
+    let now = Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true);
+    Ok(Json(api.store.put_flow_source_for_kind(
+        &path.workspace_id,
+        FlowSourceKind::Workspace,
+        &request.path,
+        &request.content,
+        &now,
+    )?))
+}
+
+async fn scoped_resolve_flow_source(
+    State(api): State<WorkspaceApi>,
+    AxumPath(path): AxumPath<ScopedWorkspacePath>,
+    Json(request): Json<FlowSourceResolveRequest>,
+) -> ApiResult<Json<ResolvedFlowSource>> {
+    validate_workspace_scope(&api, &path.workspace_id)?;
+    let resolved = match &request.selector {
+        flow::FlowSelector::Builtin { slug } => {
+            let builtin = flow::builtin_flow_source(slug)
+                .ok_or_else(|| Error::InvalidRecordId(request.selector.to_string()))?;
+            let definition = builtin.compile().map_err(|error| {
+                Error::Store(format!(
+                    "compile built-in Flow {slug:?}: {:?}",
+                    error.diagnostics
+                ))
+            })?;
+            ResolvedFlowSource {
+                selector: request.selector.clone(),
+                workspace_id: path.workspace_id,
+                flow_id: format!("builtin:{slug}"),
+                revision: builtin.revision,
+                content_digest: definition.content_digest.clone(),
+                definition,
+            }
+        }
+        flow::FlowSelector::Workspace { slug } => {
+            let source = api
+                .store
+                .get_flow_source_by_name(&path.workspace_id, FlowSourceKind::Workspace, slug)?
+                .ok_or_else(|| Error::InvalidRecordId(request.selector.to_string()))?;
+            let revision = api
+                .store
+                .get_flow_source_revision(&path.workspace_id, &source.flow_id, source.revision)?
+                .ok_or_else(|| {
+                    Error::Store(format!(
+                        "resolved Flow revision {}@{} is missing",
+                        source.flow_id, source.revision
+                    ))
+                })?;
+            ResolvedFlowSource {
+                selector: request.selector.clone(),
+                workspace_id: path.workspace_id,
+                flow_id: source.flow_id,
+                revision: source.revision,
+                content_digest: revision.content_digest,
+                definition: revision.definition,
+            }
+        }
+    };
+    Ok(Json(resolved))
+}
+
+async fn scoped_get_flow(
+    State(api): State<WorkspaceApi>,
+    AxumPath(path): AxumPath<ScopedFlowPath>,
+) -> ApiResult<Json<FlowSourceRecord>> {
+    validate_workspace_scope(&api, &path.workspace_id)?;
+    let source = api
+        .store
+        .get_flow_source(&path.workspace_id, &path.flow_id)?
+        .ok_or_else(|| Error::InvalidRecordId(path.flow_id))?;
+    Ok(Json(source))
+}
+
 async fn scoped_get_workspace(
     State(api): State<WorkspaceApi>,
     AxumPath(path): AxumPath<ScopedWorkspacePath>,
@@ -3157,7 +3284,7 @@ fn notify_ticket_recipients(
 
 fn authenticate_worker_mutation_source(
     api: &WorkspaceApi,
-    _workspace_id: &str,
+    workspace_id: &str,
     headers: &HeaderMap,
 ) -> Result<WorkerMutationSource> {
     let runtime_id = headers
@@ -3173,9 +3300,14 @@ fn authenticate_worker_mutation_source(
             Error::WorkerSourceIdentity("missing Runtime-bound Worker id".to_string())
         })?;
     let worker = RuntimeWorkerRef::new(runtime_id, worker_id);
-    api.runtime.worker(&worker).map_err(|_| {
+    let summary = api.runtime.worker(&worker).map_err(|_| {
         Error::WorkerSourceIdentity("Runtime-bound Worker identity does not exist".to_string())
     })?;
+    if summary.workspace.workspace_id.as_deref() != Some(workspace_id) {
+        return Err(Error::WorkerSourceIdentity(format!(
+            "Runtime-bound Worker is not scoped to Workspace {workspace_id}"
+        )));
+    }
     Ok(worker)
 }
 
@@ -6757,7 +6889,7 @@ async fn create_workspace_worker(
     } else {
         Some(EmbeddedWorkerInput {
             kind: EmbeddedWorkerInputKind::User,
-            content: initial_text,
+            content: initial_text.clone(),
             segments: None,
         })
     };
@@ -6776,8 +6908,9 @@ async fn create_workspace_worker(
     if resolved_working_directory.is_none() {
         reject_no_workdir_for_non_embedded_runtime(&request.runtime_id)?;
     }
+    let runtime_id = request.runtime_id.clone();
     let result = api.spawn_workspace_worker(
-        &request.runtime_id,
+        &runtime_id,
         WorkerSpawnRequest {
             requested_worker_name: Some(display_name.clone()),
             intent: WorkerSpawnIntent::WorkspaceCoding,
@@ -6798,7 +6931,7 @@ async fn create_workspace_worker(
     )?;
     Ok(Json(record_browser_worker_spawn(
         &api,
-        request.runtime_id,
+        runtime_id,
         display_name,
         selected_working_directory_id,
         result,
@@ -8949,6 +9082,7 @@ fn worker_summary_from_registry(record: &WorkerRegistryRecord) -> WorkerSummary
         workspace: WorkerWorkspaceSummary {
             visibility: "backend_registry".to_string(),
             identity: record.workspace_id.clone(),
+            workspace_id: Some(record.workspace_id.clone()),
         },
         profile: record.profile.clone(),
         implementation: WorkerImplementationSummary {
@@ -9729,7 +9863,7 @@ impl IntoResponse for ApiError {
             Error::TicketAssignmentConflict(_) | Error::WorkdirAttachmentConflict(_) => {
                 StatusCode::CONFLICT
             }
-            Error::WorkerSourceIdentity(_) => StatusCode::BAD_REQUEST,
+            Error::WorkerSourceIdentity(_) | Error::InvalidInput(_) => StatusCode::BAD_REQUEST,
             Error::InvalidRuntimeIdentifier { .. } | Error::ReservedWorkerName(_) => {
                 StatusCode::BAD_REQUEST
             }
@@ -10008,6 +10142,107 @@ mod tests {
         assert!(!serialized.contains("materialized_path"));
     }
 
+    #[tokio::test]
+    async fn flow_source_resolution_returns_immutable_workspace_and_builtin_snapshots() {
+        let workspace = tempfile::tempdir().unwrap();
+        init_clean_git_workspace(workspace.path());
+        let api = test_api(workspace.path()).await;
+        let source = r#"{
+            schema_version = 1;
+            name = "browser-flow";
+            initial = "work";
+            states = {
+                work = {
+                    instructions = "Implement and validate the requested change.";
+                    transitions = {
+                        done = { target = "done"; condition = "The work is complete."; };
+                    };
+                };
+                done = { instructions = ""; terminal = true; };
+            };
+        }"#;
+        let stored = api
+            .store
+            .put_flow_source_for_kind(
+                &api.config.workspace_id,
+                FlowSourceKind::Workspace,
+                "flows/browser-flow.dcdl",
+                source,
+                "2026-08-06T00:00:00Z",
+            )
+            .unwrap();
+
+        let Json(resolved) = scoped_resolve_flow_source(
+            State(api.clone()),
+            AxumPath(ScopedWorkspacePath {
+                workspace_id: api.config.workspace_id.clone(),
+            }),
+            Json(FlowSourceResolveRequest {
+                selector: "workspace:browser-flow".parse().unwrap(),
+            }),
+        )
+        .await
+        .unwrap();
+        assert_eq!(resolved.flow_id, stored.flow_id);
+        assert_eq!(resolved.revision, stored.revision);
+        assert_eq!(resolved.content_digest, stored.content_digest);
+        assert_eq!(resolved.definition.name, "browser-flow");
+
+        let Json(builtin) = scoped_resolve_flow_source(
+            State(api.clone()),
+            AxumPath(ScopedWorkspacePath {
+                workspace_id: api.config.workspace_id.clone(),
+            }),
+            Json(FlowSourceResolveRequest {
+                selector: "builtin:coder-review".parse().unwrap(),
+            }),
+        )
+        .await
+        .unwrap();
+        assert_eq!(builtin.definition.name, "coder-review");
+        assert_eq!(builtin.selector.to_string(), "builtin:coder-review");
+        assert_eq!(builtin.flow_id, "builtin:coder-review");
+        assert_eq!(builtin.revision, 1);
+        assert_eq!(
+            api.store
+                .list_flow_sources(&api.config.workspace_id)
+                .unwrap(),
+            vec![stored],
+            "built-in resolution must not mutate Workspace source authority",
+        );
+    }
+
+    #[tokio::test]
+    async fn worker_source_auth_rejects_cross_workspace_mutation() {
+        let workspace = tempfile::tempdir().unwrap();
+        init_clean_git_workspace(workspace.path());
+        let api = test_api(workspace.path()).await;
+        let Json(created) = create_workspace_worker(
+            State(api.clone()),
+            Json(BrowserCreateWorkerRequest {
+                runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(),
+                display_name: "Scoped Worker".to_string(),
+                profile: Some("builtin:coder".to_string()),
+                initial_text: String::new(),
+                working_directory: None,
+            }),
+        )
+        .await
+        .unwrap();
+        let mut headers = HeaderMap::new();
+        headers.insert(
+            "x-yoi-runtime-id",
+            axum::http::HeaderValue::from_str(&created.worker_ref.runtime_id).unwrap(),
+        );
+        headers.insert(
+            "x-yoi-worker-id",
+            axum::http::HeaderValue::from_str(&created.worker_ref.worker_id).unwrap(),
+        );
+        let error =
+            authenticate_worker_mutation_source(&api, "other-workspace", &headers).unwrap_err();
+        assert!(matches!(error, Error::WorkerSourceIdentity(_)));
+    }
+
     #[tokio::test]
     async fn explicit_orchestrator_launch_marks_only_the_dedicated_worker() {
         let workspace = tempfile::tempdir().unwrap();
@@ -14375,6 +14610,81 @@ mod tests {
         server.abort();
     }
 
+    #[tokio::test]
+    async fn scoped_flow_source_route_persists_compiled_dcdl() {
+        let temp = tempfile::tempdir().unwrap();
+        let app = test_app(temp.path()).await;
+        let workspace_id = test_identity().workspace_id;
+        let source = r#"{
+            schema_version = 1;
+            name = "route-flow";
+            initial = "work";
+            states = {
+                work = {
+                    instructions = "Do the work.";
+                    transitions = {
+                        done = {
+                            target = "done";
+                            condition = "The work is complete.";
+                        };
+                    };
+                };
+                done = { instructions = ""; terminal = true; };
+            };
+        }"#;
+        let response = app
+            .clone()
+            .oneshot(
+                Request::builder()
+                    .method(Method::PUT)
+                    .uri(format!("/api/w/{workspace_id}/flows"))
+                    .header(CONTENT_TYPE, "application/json")
+                    .body(Body::from(
+                        serde_json::json!({
+                            "path": "flows/route-flow.dcdl",
+                            "content": source,
+                        })
+                        .to_string(),
+                    ))
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+        assert_eq!(response.status(), StatusCode::OK);
+
+        let response = app
+            .clone()
+            .oneshot(
+                Request::builder()
+                    .method(Method::GET)
+                    .uri(format!("/api/w/{workspace_id}/flows"))
+                    .body(Body::empty())
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+        assert_eq!(response.status(), StatusCode::OK);
+
+        let response = app
+            .oneshot(
+                Request::builder()
+                    .method(Method::PUT)
+                    .uri(format!("/api/w/{workspace_id}/flows"))
+                    .header(CONTENT_TYPE, "application/json")
+                    .body(Body::from(
+                        serde_json::json!({
+                            "path": "flows/broken.dcdl",
+                            "content": "{ schema_version = 1; }",
+                        })
+                        .to_string(),
+                    ))
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+        assert_eq!(response.status(), StatusCode::BAD_REQUEST);
+    }
+
     #[tokio::test]
     async fn passkey_registration_rejects_unverified_credential_response() {
         let dir = tempfile::tempdir().unwrap();

crates/workspace-server/src/store.rs

@@ -3,8 +3,10 @@ use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
 use async_trait::async_trait;
+use flow::{CompiledFlowDefinition, FlowSourceKind, compile_flow_source};
 use rusqlite::{Connection, OptionalExtension, params};
 use serde::{Deserialize, Serialize};
+use uuid::Uuid;
 
 use worker_runtime::identity::RuntimeWorkerRef;
 
@@ -139,6 +141,16 @@ const MIGRATIONS: &[Migration] = &[
         name: "create Worker Workdir attachment reservations",
         apply: create_worker_workdir_attachment_reservations,
     },
+    Migration {
+        version: 25,
+        name: "create Flow source authority",
+        apply: create_flow_source_authority,
+    },
+    Migration {
+        version: 26,
+        name: "remove Backend-owned Flow runtime authority",
+        apply: remove_backend_flow_runtime_authority,
+    },
 ];
 
 struct Migration {
@@ -408,6 +420,31 @@ pub struct MemoryStagingResolutionRecord {
     pub resolved_at: String,
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct FlowSourceRecord {
+    pub workspace_id: String,
+    pub flow_id: String,
+    pub source_kind: FlowSourceKind,
+    pub name: String,
+    pub path: String,
+    pub content: String,
+    pub content_digest: String,
+    pub revision: u64,
+    pub created_at: String,
+    pub updated_at: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct FlowSourceRevisionRecord {
+    pub workspace_id: String,
+    pub flow_id: String,
+    pub revision: u64,
+    pub content: String,
+    pub content_digest: String,
+    pub definition: CompiledFlowDefinition,
+    pub created_at: String,
+}
+
 #[async_trait]
 pub trait ControlPlaneStore: Send + Sync {
     async fn schema_version(&self) -> Result<i64>;
@@ -417,6 +454,33 @@ pub trait ControlPlaneStore: Send + Sync {
     fn upsert_repository(&self, record: &RepositoryRecord) -> Result<()>;
     fn list_repositories(&self, workspace_id: &str) -> Result<Vec<RepositoryRecord>>;
 
+    fn put_flow_source_for_kind(
+        &self,
+        workspace_id: &str,
+        source_kind: FlowSourceKind,
+        path: &str,
+        content: &str,
+        now: &str,
+    ) -> Result<FlowSourceRecord>;
+    fn get_flow_source_by_name(
+        &self,
+        workspace_id: &str,
+        source_kind: FlowSourceKind,
+        name: &str,
+    ) -> Result<Option<FlowSourceRecord>>;
+    fn list_flow_sources(&self, workspace_id: &str) -> Result<Vec<FlowSourceRecord>>;
+    fn get_flow_source(
+        &self,
+        workspace_id: &str,
+        flow_id: &str,
+    ) -> Result<Option<FlowSourceRecord>>;
+    fn get_flow_source_revision(
+        &self,
+        workspace_id: &str,
+        flow_id: &str,
+        revision: u64,
+    ) -> Result<Option<FlowSourceRevisionRecord>>;
+
     fn upsert_objective(&self, record: &ObjectiveRecord) -> Result<()>;
     fn list_objectives(&self, workspace_id: &str, limit: usize) -> Result<Vec<ObjectiveRecord>>;
     fn get_objective(
@@ -866,6 +930,219 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
         })
     }
 
+    fn put_flow_source_for_kind(
+        &self,
+        workspace_id: &str,
+        source_kind: FlowSourceKind,
+        path: &str,
+        content: &str,
+        now: &str,
+    ) -> Result<FlowSourceRecord> {
+        if source_kind != FlowSourceKind::Workspace {
+            return Err(Error::Store(
+                "built-in Flow sources are resource authority and cannot be written to Workspace DB"
+                    .to_string(),
+            ));
+        }
+        let name = flow_source_name(path)?;
+        let definition = compile_flow_source(content).map_err(|error| {
+            Error::Store(format!(
+                "compile Flow source {path:?}: {:?}",
+                error.diagnostics
+            ))
+        })?;
+        if definition.name != name {
+            return Err(Error::Store(format!(
+                "Flow source name {:?} does not match path slug {name:?}",
+                definition.name
+            )));
+        }
+        self.with_conn(|conn| {
+            let tx = conn.unchecked_transaction()?;
+            let existing = tx
+                .query_row(
+                    r#"SELECT workspace_id, flow_id, source_kind, name, path, content,
+                              content_digest, revision, created_at, updated_at
+                       FROM flow_sources
+                       WHERE workspace_id = ?1 AND source_kind = ?2 AND name = ?3"#,
+                    params![workspace_id, source_kind.as_str(), name],
+                    read_flow_source_record,
+                )
+                .optional()?;
+            if let Some(existing) = existing {
+                if existing.content_digest == definition.content_digest {
+                    tx.commit()?;
+                    return Ok(existing);
+                }
+                let revision = existing
+                    .revision
+                    .checked_add(1)
+                    .ok_or_else(|| Error::Store("Flow source revision overflowed".to_string()))?;
+                let definition_json = serde_json::to_string(&definition)
+                    .map_err(|error| Error::Store(error.to_string()))?;
+                tx.execute(
+                    r#"INSERT INTO flow_source_revisions (
+                           workspace_id, flow_id, revision, content, content_digest,
+                           definition_json, created_at
+                       ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)"#,
+                    params![
+                        workspace_id,
+                        existing.flow_id,
+                        revision,
+                        content,
+                        definition.content_digest,
+                        definition_json,
+                        now
+                    ],
+                )?;
+                tx.execute(
+                    r#"UPDATE flow_sources
+                       SET path = ?4, content = ?5, content_digest = ?6,
+                           revision = ?7, updated_at = ?8
+                       WHERE workspace_id = ?1 AND source_kind = ?2 AND name = ?3"#,
+                    params![
+                        workspace_id,
+                        source_kind.as_str(),
+                        name,
+                        path,
+                        content,
+                        definition.content_digest,
+                        revision,
+                        now
+                    ],
+                )?;
+                tx.commit()?;
+                return Ok(FlowSourceRecord {
+                    revision,
+                    path: path.to_string(),
+                    content: content.to_string(),
+                    content_digest: definition.content_digest,
+                    updated_at: now.to_string(),
+                    ..existing
+                });
+            }
+
+            let flow_id = Uuid::now_v7().to_string();
+            let definition_json = serde_json::to_string(&definition)
+                .map_err(|error| Error::Store(error.to_string()))?;
+            tx.execute(
+                r#"INSERT INTO flow_sources (
+                       workspace_id, flow_id, source_kind, name, path, content,
+                       content_digest, revision, created_at, updated_at
+                   ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, 1, ?8, ?8)"#,
+                params![
+                    workspace_id,
+                    flow_id,
+                    source_kind.as_str(),
+                    name,
+                    path,
+                    content,
+                    definition.content_digest,
+                    now
+                ],
+            )?;
+            tx.execute(
+                r#"INSERT INTO flow_source_revisions (
+                       workspace_id, flow_id, revision, content, content_digest,
+                       definition_json, created_at
+                   ) VALUES (?1, ?2, 1, ?3, ?4, ?5, ?6)"#,
+                params![
+                    workspace_id,
+                    flow_id,
+                    content,
+                    definition.content_digest,
+                    definition_json,
+                    now
+                ],
+            )?;
+            tx.commit()?;
+            Ok(FlowSourceRecord {
+                workspace_id: workspace_id.to_string(),
+                flow_id,
+                source_kind,
+                name,
+                path: path.to_string(),
+                content: content.to_string(),
+                content_digest: definition.content_digest,
+                revision: 1,
+                created_at: now.to_string(),
+                updated_at: now.to_string(),
+            })
+        })
+    }
+
+    fn get_flow_source_by_name(
+        &self,
+        workspace_id: &str,
+        source_kind: FlowSourceKind,
+        name: &str,
+    ) -> Result<Option<FlowSourceRecord>> {
+        self.with_conn(|conn| {
+            conn.query_row(
+                r#"SELECT workspace_id, flow_id, source_kind, name, path, content,
+                          content_digest, revision, created_at, updated_at
+                   FROM flow_sources
+                   WHERE workspace_id = ?1 AND source_kind = ?2 AND name = ?3"#,
+                params![workspace_id, source_kind.as_str(), name],
+                read_flow_source_record,
+            )
+            .optional()
+            .map_err(Error::from)
+        })
+    }
+
+    fn list_flow_sources(&self, workspace_id: &str) -> Result<Vec<FlowSourceRecord>> {
+        self.with_conn(|conn| {
+            let mut statement = conn.prepare(
+                r#"SELECT workspace_id, flow_id, source_kind, name, path, content,
+                          content_digest, revision, created_at, updated_at
+                   FROM flow_sources WHERE workspace_id = ?1
+                   ORDER BY source_kind ASC, name ASC"#,
+            )?;
+            let rows = statement.query_map(params![workspace_id], read_flow_source_record)?;
+            rows.collect::<std::result::Result<Vec<_>, _>>()
+                .map_err(Error::from)
+        })
+    }
+
+    fn get_flow_source(
+        &self,
+        workspace_id: &str,
+        flow_id: &str,
+    ) -> Result<Option<FlowSourceRecord>> {
+        self.with_conn(|conn| {
+            conn.query_row(
+                r#"SELECT workspace_id, flow_id, source_kind, name, path, content,
+                          content_digest, revision, created_at, updated_at
+                   FROM flow_sources WHERE workspace_id = ?1 AND flow_id = ?2"#,
+                params![workspace_id, flow_id],
+                read_flow_source_record,
+            )
+            .optional()
+            .map_err(Error::from)
+        })
+    }
+
+    fn get_flow_source_revision(
+        &self,
+        workspace_id: &str,
+        flow_id: &str,
+        revision: u64,
+    ) -> Result<Option<FlowSourceRevisionRecord>> {
+        self.with_conn(|conn| {
+            conn.query_row(
+                r#"SELECT workspace_id, flow_id, revision, content, content_digest,
+                          definition_json, created_at
+                   FROM flow_source_revisions
+                   WHERE workspace_id = ?1 AND flow_id = ?2 AND revision = ?3"#,
+                params![workspace_id, flow_id, revision],
+                read_flow_source_revision_record,
+            )
+            .optional()
+            .map_err(Error::from)
+        })
+    }
+
     fn upsert_objective(&self, record: &ObjectiveRecord) -> Result<()> {
         self.with_conn(|conn| {
             conn.execute(
@@ -2660,6 +2937,79 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
     }
 }
 
+fn flow_source_name(path: &str) -> Result<String> {
+    let file_name = path
+        .rsplit('/')
+        .next()
+        .filter(|value| !value.is_empty())
+        .ok_or_else(|| Error::Store("Flow source path has no file name".to_string()))?;
+    let name = file_name
+        .strip_suffix(".dcdl")
+        .filter(|value| !value.is_empty())
+        .ok_or_else(|| Error::Store("Flow source path must end in .dcdl".to_string()))?;
+    flow::FlowSelector::builtin(name)
+        .map_err(|error| Error::Store(format!("invalid Flow source slug: {error}")))?;
+    Ok(name.to_string())
+}
+
+fn read_flow_source_record(row: &rusqlite::Row<'_>) -> rusqlite::Result<FlowSourceRecord> {
+    let source_kind = match row.get::<_, String>(2)?.as_str() {
+        "builtin" => FlowSourceKind::Builtin,
+        "workspace" => FlowSourceKind::Workspace,
+        other => {
+            return Err(rusqlite::Error::FromSqlConversionFailure(
+                2,
+                rusqlite::types::Type::Text,
+                format!("invalid Flow source kind {other:?}").into(),
+            ));
+        }
+    };
+    let revision = row.get::<_, i64>(7)?;
+    Ok(FlowSourceRecord {
+        workspace_id: row.get(0)?,
+        flow_id: row.get(1)?,
+        source_kind,
+        name: row.get(3)?,
+        path: row.get(4)?,
+        content: row.get(5)?,
+        content_digest: row.get(6)?,
+        revision: u64::try_from(revision).map_err(|error| {
+            rusqlite::Error::FromSqlConversionFailure(
+                7,
+                rusqlite::types::Type::Integer,
+                Box::new(error),
+            )
+        })?,
+        created_at: row.get(8)?,
+        updated_at: row.get(9)?,
+    })
+}
+
+fn read_flow_source_revision_record(
+    row: &rusqlite::Row<'_>,
+) -> rusqlite::Result<FlowSourceRevisionRecord> {
+    let revision = row.get::<_, i64>(2)?;
+    let definition_json = row.get::<_, String>(5)?;
+    let definition = serde_json::from_str(&definition_json).map_err(|error| {
+        rusqlite::Error::FromSqlConversionFailure(5, rusqlite::types::Type::Text, Box::new(error))
+    })?;
+    Ok(FlowSourceRevisionRecord {
+        workspace_id: row.get(0)?,
+        flow_id: row.get(1)?,
+        revision: u64::try_from(revision).map_err(|error| {
+            rusqlite::Error::FromSqlConversionFailure(
+                2,
+                rusqlite::types::Type::Integer,
+                Box::new(error),
+            )
+        })?,
+        content: row.get(3)?,
+        content_digest: row.get(4)?,
+        definition,
+        created_at: row.get(6)?,
+    })
+}
+
 fn read_workspace_record(row: &rusqlite::Row<'_>) -> rusqlite::Result<WorkspaceRecord> {
     Ok(WorkspaceRecord {
         workspace_id: row.get(0)?,
@@ -3623,6 +3973,52 @@ CREATE TABLE IF NOT EXISTS __yoi_schema_migrations (
     Ok(())
 }
 
+fn create_flow_source_authority(conn: &Connection) -> Result<()> {
+    conn.execute_batch(
+        r#"
+CREATE TABLE flow_sources (
+    workspace_id TEXT NOT NULL,
+    flow_id TEXT NOT NULL,
+    source_kind TEXT NOT NULL CHECK (source_kind IN ('builtin', 'workspace')),
+    name TEXT NOT NULL,
+    path TEXT NOT NULL,
+    content TEXT NOT NULL,
+    content_digest TEXT NOT NULL,
+    revision INTEGER NOT NULL CHECK (revision > 0),
+    created_at TEXT NOT NULL,
+    updated_at TEXT NOT NULL,
+    PRIMARY KEY (workspace_id, flow_id),
+    UNIQUE (workspace_id, source_kind, name),
+    FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE
+);
+CREATE TABLE flow_source_revisions (
+    workspace_id TEXT NOT NULL,
+    flow_id TEXT NOT NULL,
+    revision INTEGER NOT NULL CHECK (revision > 0),
+    content TEXT NOT NULL,
+    content_digest TEXT NOT NULL,
+    definition_json TEXT NOT NULL,
+    created_at TEXT NOT NULL,
+    PRIMARY KEY (workspace_id, flow_id, revision),
+    FOREIGN KEY (workspace_id, flow_id)
+        REFERENCES flow_sources(workspace_id, flow_id) ON DELETE CASCADE
+);
+"#,
+    )?;
+    Ok(())
+}
+
+fn remove_backend_flow_runtime_authority(conn: &Connection) -> Result<()> {
+    conn.execute_batch(
+        r#"
+DROP TABLE IF EXISTS flow_events;
+DROP TABLE IF EXISTS flow_transition_attempts;
+DROP TABLE IF EXISTS flow_instances;
+"#,
+    )?;
+    Ok(())
+}
+
 fn current_schema_version(conn: &Connection) -> Result<i64> {
     conn.query_row(
         "SELECT COALESCE(MAX(version), 0) FROM __yoi_schema_migrations",
@@ -4182,17 +4578,54 @@ CREATE TABLE ticket_worker_links (ticket_id TEXT, worker_ref_key TEXT);
 
         apply_migrations(&conn).unwrap();
 
-        assert_eq!(current_schema_version(&conn).unwrap(), 24);
+        assert_eq!(current_schema_version(&conn).unwrap(), 26);
         assert!(table_exists(&conn, "worker_workdir_attachment_reservations").unwrap());
     }
 
+    #[test]
+    fn schema_v26_removes_legacy_backend_flow_runtime_tables() {
+        let conn = Connection::open_in_memory().unwrap();
+        configure_sqlite(&conn).unwrap();
+        for migration in MIGRATIONS
+            .iter()
+            .filter(|migration| migration.version <= 25)
+        {
+            let tx = conn.unchecked_transaction().unwrap();
+            (migration.apply)(&tx).unwrap();
+            tx.execute(
+                "INSERT INTO __yoi_schema_migrations (version, name) VALUES (?1, ?2)",
+                params![migration.version, migration.name],
+            )
+            .unwrap();
+            tx.commit().unwrap();
+        }
+        conn.execute_batch(
+            r#"
+CREATE TABLE flow_instances (instance_id TEXT PRIMARY KEY);
+CREATE TABLE flow_transition_attempts (attempt_id TEXT PRIMARY KEY);
+CREATE TABLE flow_events (event_id TEXT PRIMARY KEY);
+"#,
+        )
+        .unwrap();
+        assert_eq!(current_schema_version(&conn).unwrap(), 25);
+
+        apply_migrations(&conn).unwrap();
+
+        assert_eq!(current_schema_version(&conn).unwrap(), 26);
+        assert!(table_exists(&conn, "flow_sources").unwrap());
+        assert!(table_exists(&conn, "flow_source_revisions").unwrap());
+        assert!(!table_exists(&conn, "flow_instances").unwrap());
+        assert!(!table_exists(&conn, "flow_transition_attempts").unwrap());
+        assert!(!table_exists(&conn, "flow_events").unwrap());
+    }
+
     #[tokio::test]
     async fn migrates_sqlite_and_preserves_workspace_record() {
         let dir = tempfile::tempdir().unwrap();
         let db = dir.path().join("control-plane.sqlite");
         let store = SqliteWorkspaceStore::open(&db).unwrap();
 
-        assert_eq!(store.schema_version().await.unwrap(), 24);
+        assert_eq!(store.schema_version().await.unwrap(), 26);
         assert!(
             !store
                 .with_conn(|conn| table_exists(conn, "worker_workspace_credentials"))
@@ -4209,13 +4642,88 @@ CREATE TABLE ticket_worker_links (ticket_id TEXT, worker_ref_key TEXT);
         store.upsert_workspace(&record).await.unwrap();
 
         let reopened = SqliteWorkspaceStore::open(&db).unwrap();
-        assert_eq!(reopened.schema_version().await.unwrap(), 24);
+        assert_eq!(reopened.schema_version().await.unwrap(), 26);
         assert_eq!(
             reopened.get_workspace("local-dev").await.unwrap(),
             Some(record)
         );
     }
 
+    #[tokio::test]
+    async fn workspace_flow_sources_keep_revisions_and_builtins_stay_resources() {
+        let dir = tempfile::tempdir().unwrap();
+        let store = SqliteWorkspaceStore::open(dir.path().join("server.db")).unwrap();
+        store
+            .upsert_workspace(&WorkspaceRecord {
+                workspace_id: "workspace-a".to_string(),
+                owner_account_id: None,
+                display_name: "Workspace A".to_string(),
+                state: "active".to_string(),
+                created_at: "2026-08-06T00:00:00Z".to_string(),
+                updated_at: "2026-08-06T00:00:00Z".to_string(),
+            })
+            .await
+            .unwrap();
+        let workspace_source = r#"{
+            schema_version = 1;
+            name = "coder-review";
+            initial = "work";
+            states = {
+                work = {
+                    instructions = "Workspace revision one.";
+                    transitions = { done = { target = "done"; condition = "Done."; }; };
+                };
+                done = { instructions = ""; terminal = true; };
+            };
+        }"#;
+        let workspace = store
+            .put_flow_source_for_kind(
+                "workspace-a",
+                FlowSourceKind::Workspace,
+                "flows/coder-review.dcdl",
+                workspace_source,
+                "2026-08-06T00:00:01Z",
+            )
+            .unwrap();
+        let builtin = flow::builtin_flow_source("coder-review").unwrap();
+        assert_eq!(builtin.slug, workspace.name);
+        assert!(
+            store
+                .put_flow_source_for_kind(
+                    "workspace-a",
+                    FlowSourceKind::Builtin,
+                    builtin.path,
+                    builtin.content,
+                    "2026-08-06T00:00:02Z",
+                )
+                .is_err()
+        );
+        assert_eq!(
+            store.list_flow_sources("workspace-a").unwrap(),
+            vec![workspace.clone()]
+        );
+
+        let revision_two =
+            workspace_source.replace("Workspace revision one.", "Workspace revision two.");
+        let updated = store
+            .put_flow_source_for_kind(
+                "workspace-a",
+                FlowSourceKind::Workspace,
+                "flows/coder-review.dcdl",
+                &revision_two,
+                "2026-08-06T00:00:03Z",
+            )
+            .unwrap();
+        assert_eq!(updated.flow_id, workspace.flow_id);
+        assert_eq!(updated.revision, 2);
+        let pinned = store
+            .get_flow_source_revision("workspace-a", &workspace.flow_id, 1)
+            .unwrap()
+            .unwrap();
+        assert_eq!(pinned.content, workspace_source);
+        assert_eq!(pinned.definition.name, "coder-review");
+    }
+
     #[tokio::test]
     async fn ticket_worker_assignment_replaces_current_and_preserves_audit_history() {
         let dir = tempfile::tempdir().unwrap();
@@ -4681,7 +5189,7 @@ CREATE TABLE ticket_worker_links (ticket_id TEXT, worker_ref_key TEXT);
         .unwrap();
 
         let store = SqliteWorkspaceStore::from_connection(conn).unwrap();
-        assert_eq!(store.schema_version().await.unwrap(), 24);
+        assert_eq!(store.schema_version().await.unwrap(), 26);
 
         store
             .with_conn(|conn| {
@@ -4870,7 +5378,7 @@ CREATE TABLE ticket_assignment_operations (
     #[tokio::test]
     async fn repository_records_round_trip() {
         let store = SqliteWorkspaceStore::in_memory().unwrap();
-        assert_eq!(store.schema_version().await.unwrap(), 24);
+        assert_eq!(store.schema_version().await.unwrap(), 26);
         let workspace = WorkspaceRecord {
             workspace_id: "local-dev".to_string(),
             owner_account_id: None,
@@ -4908,7 +5416,7 @@ CREATE TABLE ticket_assignment_operations (
     #[tokio::test]
     async fn memory_authority_records_round_trip_and_close_staging() {
         let store = SqliteWorkspaceStore::in_memory().unwrap();
-        assert_eq!(store.schema_version().await.unwrap(), 24);
+        assert_eq!(store.schema_version().await.unwrap(), 26);
         let workspace = WorkspaceRecord {
             workspace_id: "local-dev".to_string(),
             owner_account_id: None,
@@ -5156,7 +5664,7 @@ CREATE TABLE ticket_assignment_operations (
     #[tokio::test]
     async fn account_and_login_records_round_trip() {
         let store = SqliteWorkspaceStore::in_memory().unwrap();
-        assert_eq!(store.schema_version().await.unwrap(), 24);
+        assert_eq!(store.schema_version().await.unwrap(), 26);
         let now = "2026-07-22T00:00:00Z".to_string();
         let account = AccountRecord {
             account_id: "acct-user-alice".to_string(),