flow_application/

lib.rs

use chrono::{DateTime, Utc};
use flow_adapter_cli::CliAdapter;
use flow_adapter_fs::FsAdapter;
use flow_adapter_shell::ShellAdapter;
use flow_adapter_utility::UtilityAdapter;
use flow_adapter_ai::{
    ClaudeProvider, CloudAiProvider, CloudAiRegistry, DeepSeekProvider, GeminiProvider,
    LocalOpenAiProvider, NvidiaProvider, OpenAiProvider,
};
use flow_domain::execution::{ExecutionRequest, ExecutionResult, ExecutionStatus};
use flow_domain::graph::{FlowEdge, FlowGraph, FlowNode, Position};
use flow_execution::{
    AdapterRegistry, CapturingSink, EventSink, ExecutionEvent,
    ExecutionSummary, Executor, MockAdapter, MultiSink, ProposedEdit,
};
use flow_security::{
    CredentialKind, CredentialResolver, CredentialStore, EnvFallbackResolver,
    KeyringCredentialStore, PiiSanitizer,
};
use flow_storage::{ExecutionRecord, ExecutionStepRecord, InterceptionRecord, ScheduleRecord, Store};
use std::path::PathBuf;
use std::sync::Arc;
use thiserror::Error;
use uuid::Uuid;

pub mod ansible_import;
pub mod cli_tools;
pub mod connections;
pub mod node_hub;
pub mod nodes;
pub mod dsl_semantics;
pub mod hub;
pub mod install;
// The local model-server lifecycle lives in its own crate so the CLI can reuse
// it; re-exported under the original `llm_server` path to keep every call site
// (here and in the desktop) unchanged.
pub use flow_models_server as llm_server;
pub mod scheduler;
pub mod settings;
pub mod template_hub;
pub mod templates;
pub use connections::{ConnectionStore, StoreError as ConnectionStoreError};
pub use flow_execution::{ConnectionError, ConnectionLookup, FlowWarning, ResolvedZoweConnection};
pub use install::{InstallProgress, ProgressCallback};
pub use settings::{Settings, SettingsPatch, SettingsStore};
pub use templates::TemplateRecord;

#[derive(Debug, Error)]
pub enum AppError {
    #[error("graph not found: {0}")]
    GraphNotFound(String),
    #[error("validation failed: {0}")]
    Validation(String),
    #[error("ai invocation failed: {0}")]
    AiInvocation(String),
    #[error("execution failed: {0}")]
    Execution(String),
    #[error("storage error: {0}")]
    Storage(String),
    #[error("settings error: {0}")]
    Settings(String),
    #[error("template error: {0}")]
    Template(String),
    #[error("credential error: {0}")]
    Credential(String),
    #[error("dsl parse error at line {line}:{col}: {message}")]
    Dsl {
        line: u32,
        col: u32,
        message: String,
    },
}

/// Outcome of a coding-agent turn run via [`FlowApp::run_agent_turn`]: the run
/// summary plus the filesystem edits the agent proposed (staged, not yet
/// applied) for the IDE to review and apply.
#[derive(Debug, Clone, serde::Serialize)]
pub struct AgentTurn {
    pub summary: ExecutionSummary,
    pub edits: Vec<ProposedEdit>,
}

pub struct FlowApp {
    sanitizer: Arc<PiiSanitizer>,
    adapters: Arc<AdapterRegistry>,
    cloud_providers: Vec<Arc<dyn CloudAiProvider>>,
    credentials: Arc<dyn CredentialStore>,
    resolver: Arc<dyn CredentialResolver>,
    connections: ConnectionStore,
    settings: SettingsStore,
    store: Store,
    llm_server: crate::llm_server::LlmServerHandle,
    /// Fan-out target for streaming LLM tokens. Installed by the host
    /// (Tauri main wires a `TauriStreamSink` that broadcasts
    /// `flow:llm_token`); defaults to a no-op for headless tests.
    ///
    /// Wrapped in `RwLock<Arc<...>>` because Tauri manages `FlowApp` as
    /// shared state (`&FlowApp` only), but the sink needs to be
    /// installed from the `setup` closure after the `AppHandle` exists.
    /// Callers fetch a clone of the current Arc via `stream_sink()` so
    /// long-running futures aren't holding the lock.
    stream_sink: std::sync::RwLock<Arc<dyn flow_adapter_ai::LlmStreamSink>>,
    /// Session working memory backing `{{memory.<key>}}`. Owned here (not on
    /// the per-run `Executor`) so it survives across `execute_graph` calls -
    /// i.e. across re-plan iterations within an agent session. Cleared on app
    /// restart; reset mid-session via [`FlowApp::clear_working_memory`].
    working_memory: flow_execution::WorkingMemory,
    /// Active runs keyed by execution id, each with its own pause/resume/cancel
    /// control so multiple flows (e.g. several canvas tabs) can run and be
    /// steered independently. A fresh control is inserted by
    /// [`FlowApp::execute_graph_with_id`] for the duration of a run and removed
    /// on completion. Steered per-run via [`FlowApp::pause_run`] /
    /// [`resume_run`] / [`cancel_run`], or across all runs via the `*_all_runs`
    /// helpers (used by headless runners that track a single run).
    runs: std::sync::Mutex<std::collections::HashMap<String, flow_execution::SharedRunControl>>,
    /// Cancellation flag for an in-flight Hub model download. Set by
    /// [`FlowApp::cancel_download`], polled by the streaming loop in
    /// `hub::download`. One flag suffices - the frontend allows a single
    /// download at a time.
    download_cancel: Arc<std::sync::atomic::AtomicBool>,
    /// Forces the per-step destructive-action confirmation gate off for this
    /// `FlowApp`, regardless of the persisted setting. Set by headless runners
    /// (the CLI / TUI) via [`FlowApp::disable_confirmation_gate`] since they
    /// have no way to deliver a confirmation - a destructive node would block
    /// forever. Desktop / server leave it `false` so the setting applies.
    confirm_gate_disabled: std::sync::atomic::AtomicBool,
    /// How this host derives a flow's default execution workspace when no
    /// per-flow override is stored. Desktop/server use `Scratch`; flow-cli sets
    /// `Fixed` to the launch dir via [`FlowApp::with_workspace_base`].
    workspace_base: WorkspaceBase,
}

/// Where a flow's `shell` / `cli` / `fs` nodes run, before any explicit
/// per-node `cwd` / `workspaceRoot`. The default is edition-specific so the
/// shared core serves desktop, server, and CLI from one resolution path.
#[derive(Debug, Clone)]
pub enum WorkspaceBase {
    /// Per-flow scratch at `<home>/.flow-studio/work/<flow_id>/`. HOME-based
    /// (not redirected by `FLOW_STUDIO_DIR`), matching the shell scratch.
    Scratch,
    /// One directory shared by every flow - the flow-cli launch dir (or its
    /// `--workspace`).
    Fixed(PathBuf),
}

impl Default for WorkspaceBase {
    fn default() -> Self {
        WorkspaceBase::Scratch
    }
}

impl WorkspaceBase {
    fn resolve(&self, flow_id: &str) -> PathBuf {
        match self {
            WorkspaceBase::Scratch => FlowApp::home_dir()
                .join(".flow-studio")
                .join("work")
                .join(flow_id),
            WorkspaceBase::Fixed(path) => path.clone(),
        }
    }
}

/// Expand a leading `~/` against `$HOME`; other paths pass through unchanged.
fn expand_home(raw: &str) -> PathBuf {
    if let Some(rest) = raw.strip_prefix("~/") {
        if let Some(home) = std::env::var_os("HOME") {
            return PathBuf::from(home).join(rest);
        }
    }
    PathBuf::from(raw)
}

impl Default for FlowApp {
    fn default() -> Self {
        Self::new()
    }
}

/// System prompt used when routing the Generate tab through a cloud provider.
///
/// Sourced from the shared `docs/dsl/spec_compiled.md` artifact (which flow-ml
/// also syncs into the flow-graph-generator's training prompt). That file is
/// regenerated by `scripts/build-dsl-spec.sh` (driven by this crate's
/// `build.rs` whenever any source under `docs/dsl/` or any adapter
/// `descriptor()` changes), so editing the spec without re-running the
/// script is impossible: the build-time script call refreshes
/// `spec_compiled.md` first.
///
/// The artifact begins with a one-paragraph "emit DSL only - no markdown
/// fences, no preamble" header. `extract_dsl_text` below still strips an
/// outer fence as a belt-and-braces measure for general-purpose cloud
/// models that occasionally ignore the instruction.
const DSL_GENERATION_SYSTEM_PROMPT: &str = include_str!("../../../docs/dsl/spec_compiled.md");

/// Outcome of a hybrid flow-graph generation. Carries the DSL text plus a
/// machine-readable record of which path produced it so the UI can
/// render a "local" / "cloud (fallback)" tag without re-deriving the
/// information.
///
/// Serialised to camelCase (via the Tauri command layer) so the
/// frontend's `generateDslAuto` wrapper consumes
/// `{ dsl, source, fallbackReason }`.
#[derive(Debug, Clone, serde::Serialize)]
pub struct DslGenerationOutcome {
    pub dsl: String,
    pub source: DslSource,
    pub fallback_reason: Option<String>,
    /// Optional natural-language summary of the generated DSL. Produced by
    /// the agentic path via a second chat call; non-agentic generation
    /// leaves this `None`. The frontend falls back to a deterministic
    /// client-side synthesis when this is missing.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub plan: Option<String>,
}

/// Safety gate for the autonomous agentic loop: the hard ceiling on
/// generate→run→observe→re-plan cycles. The loop stops the moment a run has
/// zero failures; this bounds the unhappy path so the agent can't spin
/// indefinitely (and can't rack up unbounded model/compute cost).
pub const MAX_AGENTIC_ITERATIONS: u32 = 5;

/// Result of an autonomous agentic loop ([`FlowApp::agentic_run_loop`]).
/// Serialised camelCase for the frontend, which loads `finalDsl` onto the
/// canvas and renders the per-iteration `steps` as a run log.
#[derive(Debug, Clone, serde::Serialize)]
#[serde(rename_all = "camelCase")]
pub struct AgenticLoopSummary {
    /// Number of generate→run cycles actually performed (1..=MAX).
    pub iterations: u32,
    /// `"succeeded"` if a run reached zero failures; `"exhausted"` if the
    /// iteration cap was hit without converging; `"budget_exhausted"` if the
    /// wall-clock ceiling (`settings.max_agentic_seconds`) elapsed first.
    pub status: String,
    /// DSL from the last iteration (the converged graph on success, or the
    /// final attempt on exhaustion).
    pub final_dsl: String,
    pub final_plan: Option<String>,
    pub steps: Vec<AgenticLoopStep>,
}

/// One iteration of the autonomous loop: the run outcome plus the feedback
/// that was fed into the next attempt (`None` on the converged iteration).
#[derive(Debug, Clone, serde::Serialize)]
#[serde(rename_all = "camelCase")]
pub struct AgenticLoopStep {
    pub iteration: u32,
    pub run_status: String,
    pub succeeded: usize,
    pub failed: usize,
    pub skipped: usize,
    pub observations: Option<String>,
}

/// Host machine snapshot for the Model Hub device-compatibility check. RAM/disk
/// are in binary GB (GiB) so they compare directly against the catalog's
/// `minRamGb` / `minDiskGb`.
#[derive(Debug, Clone, serde::Serialize)]
#[serde(rename_all = "camelCase")]
pub struct SystemInfo {
    /// `macos` | `linux` | `windows` (from `std::env::consts::OS`).
    pub os: String,
    /// `aarch64` | `x86_64` | … (from `std::env::consts::ARCH`).
    pub arch: String,
    pub total_ram_gb: f64,
    pub available_ram_gb: f64,
    pub free_disk_gb: f64,
    /// Logical CPU count (drives the threads slider max).
    pub cpu_count: u32,
}

/// Tagged enum recording which side of the hybrid Auto path produced the
/// DSL. The internally-tagged JSON form (`{ "kind": "local"|"cloud", ... }`)
/// is what the frontend's `ModelChoice` discriminator expects to receive.
#[derive(Debug, Clone, serde::Serialize)]
#[serde(tag = "kind", rename_all = "lowercase")]
pub enum DslSource {
    Local {
        #[serde(rename = "modelId")]
        model_id: String,
    },
    Cloud {
        provider: String,
        #[serde(rename = "modelId")]
        model_id: String,
    },
}

/// Robustly extract the DSL body from a cloud model's response.
///
/// The system prompt instructs the model to emit DSL only, but Claude / GPT-4
/// occasionally still wrap the response in a code fence or prefix a one-line
/// preamble. We trim leading whitespace, strip a leading fence (` ``` ` with
/// optional language tag), and strip a trailing fence if matched.
fn extract_dsl_text(raw: &str) -> String {
    let trimmed = raw.trim();
    let with_fence_stripped = if trimmed.starts_with("```") {
        let after_first = trimmed.split_once('\n').map(|x| x.1).unwrap_or("");
        match after_first.rfind("```") {
            Some(idx) => after_first[..idx].trim_end().to_string(),
            None => after_first.to_string(),
        }
    } else {
        trimmed.to_string()
    };
    escape_newlines_in_strings(with_fence_stripped.trim())
}

/// Escape raw newlines that appear *inside* double-quoted string literals.
///
/// Cloud models frequently emit multi-line file `content` with real line breaks
/// inside the quotes, which the lexer rejects as `unterminated string` - and no
/// amount of corrective prompting reliably stops some models from doing it. This
/// normalizes those breaks to `\n` so the DSL parses, while leaving everything
/// outside strings untouched. Backslash escapes are honored (`\"` does not close
/// a string; `\\` is an escaped backslash), and `#` outside a string opens a
/// line comment in which quotes are literal. A genuinely unterminated string
/// (no closing quote before EOF) still runs to EOF and is reported as an error,
/// so real mistakes aren't masked.
fn escape_newlines_in_strings(src: &str) -> String {
    let mut out = String::with_capacity(src.len());
    let mut in_string = false;
    let mut in_comment = false;
    let mut escaped = false;
    for ch in src.chars() {
        if in_comment {
            out.push(ch);
            if ch == '\n' {
                in_comment = false;
            }
            continue;
        }
        if in_string {
            if escaped {
                out.push(ch);
                escaped = false;
            } else if ch == '\\' {
                out.push(ch);
                escaped = true;
            } else if ch == '"' {
                out.push(ch);
                in_string = false;
            } else if ch == '\n' {
                out.push_str("\\n");
            } else if ch == '\r' {
                // Drop a bare CR inside a string; a following LF becomes `\n`.
            } else {
                out.push(ch);
            }
            continue;
        }
        match ch {
            '"' => {
                in_string = true;
                out.push(ch);
            }
            '#' => {
                in_comment = true;
                out.push(ch);
            }
            _ => out.push(ch),
        }
    }
    out
}

/// One round-trip through a cloud provider with the flow-graph-generator system
/// prompt. Pulled out of `generate_dsl_via_cloud` so the parse-then-retry
/// path can reuse it without duplicating the `CloudAiRequest`
/// construction or error mapping.
async fn cloud_call_once(
    provider: &dyn CloudAiProvider,
    model: &str,
    system: &str,
    user: &str,
    max_tokens: Option<u32>,
    temperature: Option<f32>,
    api_key: &str,
    base_url: Option<String>,
    sink: &dyn flow_adapter_ai::LlmStreamSink,
) -> Result<(String, String), AppError> {
    // Visible at the default `info` level so the server terminal shows an AI
    // provider was actually called for generation (round 1 and each retry).
    tracing::info!(
        provider = provider.name(),
        model,
        "AI provider call for Flow DSL generation"
    );
    let req = flow_adapter_ai::CloudAiRequest {
        model: model.to_string(),
        prompt: user.to_string(),
        max_tokens,
        temperature,
        api_key: api_key.to_string(),
        system: Some(system.to_string()),
        // Cloud providers ignore this (fixed endpoint); the `local`
        // provider reads it as its on-device server address.
        base_url,
        // Flow-graph generation doesn't expose advanced sampling knobs and the
        // response is structured DSL, so leave temperature and the rest at
        // server defaults.
        stream: true,
        call_id: Some(format!("dsl-{}", uuid::Uuid::new_v4())),
        ..Default::default()
    };
    let resp = provider
        .invoke_stream(&req, sink)
        .await
        .map_err(|e| AppError::AiInvocation(e.to_string()))?;
    // Hand back the finish/stop reason too: a response cut off at the token
    // limit needs different handling (a clear error) than one that finished
    // but wrote bad DSL (a corrective retry).
    Ok((resp.text, resp.finish_reason))
}

/// Finish/stop reasons that mean the model was cut off at the token limit
/// rather than completing its answer. OpenAI-compatible servers report
/// `"length"`, Anthropic `"max_tokens"`, Gemini `"MAX_TOKENS"`. A truncated
/// response almost always ends mid-string, so the DSL won't parse - and
/// re-prompting at the same budget just reproduces it.
fn is_truncated(finish_reason: &str) -> bool {
    matches!(
        finish_reason,
        "length" | "max_tokens" | "MAX_TOKENS" | "model_length"
    )
}

/// Clear, actionable error for a generation cut off at the token limit, in
/// place of the misleading `unterminated string` the truncation produces.
fn truncation_error(
    provider: &str,
    model: &str,
    max_tokens: Option<u32>,
    last_reason: &str,
) -> AppError {
    let budget = max_tokens
        .map(|t| format!(" ({t}-token budget)"))
        .unwrap_or_default();
    tracing::warn!(
        provider,
        model,
        %last_reason,
        "Flow DSL generation hit the token limit (response truncated)"
    );
    AppError::AiInvocation(format!(
        "{provider} model's response was cut off at the token limit{budget} before the \
         flow was complete - the generated flow is too large (a file `content` value was \
         truncated mid-string). Try a more focused request, fewer or smaller inline files, \
         or a higher generation token budget."
    ))
}

/// Cap a (possibly large) generated DSL blob before it goes into a log line,
/// so a failed generation doesn't dump kilobytes into the server terminal.
/// Truncates on a char boundary and notes how many bytes were dropped.
fn truncate_for_log(s: &str) -> String {
    const MAX: usize = 1200;
    if s.len() <= MAX {
        return s.to_string();
    }
    let mut end = MAX;
    while !s.is_char_boundary(end) {
        end -= 1;
    }
    format!("{}... (+{} more bytes)", &s[..end], s.len() - end)
}

/// Render the offending source line plus a caret under `col` (both 1-based),
/// so the corrective-retry prompt shows the model exactly where parsing broke
/// instead of an abstract `line:col`. Empty string if the line is out of range.
fn offending_line_caret(src: &str, line: u32, col: u32) -> String {
    let Some(text) = src.lines().nth(line.saturating_sub(1) as usize) else {
        return String::new();
    };
    let nchars = text.chars().count();
    let caret_at = (col.saturating_sub(1) as usize).min(nchars);
    let pad = " ".repeat(caret_at);
    format!("{text}\n{pad}^")
}

/// Build the corrective user prompt sent on the retry leg. Quotes the
/// previous (broken) DSL between sentinel markers, reports the parser's
/// `line:col` + message, then restates the original request and asks
/// for corrected DSL only - same "no commentary" instruction the
/// system prompt already gives, repeated here so the model can't
/// fall back to apologetic chatter.
///
/// Sentinels (rather than a ```` ```flow ```` fence) avoid an ambiguous
/// prompt when the previous attempt itself contained a triple-backtick
/// run: the closing fence would land in the wrong place and confuse
/// the model about where its prior output ended.
fn build_retry_prompt(
    original: &str,
    previous_dsl: &str,
    parse_err: &flow_dsl::DslError,
) -> String {
    let hint = parse_error_hint(&parse_err.message)
        .map(|h| format!("\n{h}\n"))
        .unwrap_or_default();
    let pointer = offending_line_caret(previous_dsl, parse_err.line, parse_err.col);
    let pointer_block = if pointer.is_empty() {
        String::new()
    } else {
        format!("\nThe error is here:\n{pointer}\n")
    };
    format!(
        "Your previous attempt failed to parse:\n\
         \n\
         <<<BEGIN_PREVIOUS_DSL>>>\n\
         {previous_dsl}\n\
         <<<END_PREVIOUS_DSL>>>\n\
         \n\
         Parser error: line {line}:{col} - {message}\n\
         {pointer_block}\
         {hint}\
         \n\
         Original request:\n\
         {original}\n\
         \n\
         Return ONLY corrected Flow DSL. No commentary, no fences.",
        line = parse_err.line,
        col = parse_err.col,
        message = parse_err.message,
    )
}

fn build_semantic_retry_prompt(original: &str, previous_dsl: &str, issues: &[String]) -> String {
    let bullets = issues
        .iter()
        .map(|i| format!("- {i}"))
        .collect::<Vec<_>>()
        .join("\n");
    format!(
        "Your previous DSL parsed but violates Flow semantics:\n\
         \n\
         {bullets}\n\
         \n\
         <<<BEGIN_PREVIOUS_DSL>>>\n\
         {previous_dsl}\n\
         <<<END_PREVIOUS_DSL>>>\n\
         \n\
         Original request:\n\
         {original}\n\
         \n\
         Return ONLY corrected Flow DSL. No commentary, no fences. \
         Use `flow \"<name>\" v1` as the header (not v1.0.0). \
         Do not invent git commits - read the real history from the repo with a shell action.",
    )
}

/// Parse, repair common model mistakes, validate semantics, return canonical DSL.
/// Turn a finished run's captured events + summary into a compact,
/// model-facing feedback string for the next re-plan iteration. Lists the
/// failed and skipped nodes with their reasons (deduped to the latest state
/// per node, since a bounded loop can touch a node more than once).
/// Sum the token usage of every AI node that succeeded in a run. The executor's
/// `ai_generate_envelope` stamps `input_tokens`/`output_tokens` onto each AI
/// node's output, so we read them back from the captured `NodeSucceeded` events.
/// Used by `agentic_run_loop`'s token budget; non-AI nodes carry no token fields
/// and contribute zero.
fn sum_ai_tokens(events: &[ExecutionEvent]) -> u64 {
    let mut total = 0u64;
    for ev in events {
        if let ExecutionEvent::NodeSucceeded { output, .. } = ev {
            let i = output
                .get("input_tokens")
                .and_then(|v| v.as_u64())
                .unwrap_or(0);
            let o = output
                .get("output_tokens")
                .and_then(|v| v.as_u64())
                .unwrap_or(0);
            total = total.saturating_add(i).saturating_add(o);
        }
    }
    total
}

fn build_loop_observations(events: &[ExecutionEvent], summary: &ExecutionSummary) -> String {
    use std::collections::BTreeMap;
    let mut lines = vec![format!(
        "The workflow ran: {} succeeded, {} failed, {} skipped (status: {}).",
        summary.succeeded, summary.failed, summary.skipped, summary.status
    )];
    let mut failures: BTreeMap<&str, &str> = BTreeMap::new();
    let mut skips: BTreeMap<&str, &str> = BTreeMap::new();
    for e in events {
        match e {
            ExecutionEvent::NodeFailed { node_id, error, .. } => {
                failures.insert(node_id, error);
            }
            ExecutionEvent::NodeSkipped { node_id, reason, .. } => {
                skips.insert(node_id, reason);
            }
            _ => {}
        }
    }
    if !failures.is_empty() {
        lines.push("Failures:".into());
        lines.extend(failures.iter().map(|(n, e)| format!("- node `{n}` failed: {e}")));
    }
    if !skips.is_empty() {
        lines.push("Skipped (never ran):".into());
        lines.extend(skips.iter().map(|(n, r)| format!("- node `{n}` skipped: {r}")));
    }
    lines.join("\n")
}

fn validate_generated_dsl(dsl: &str) -> Result<String, String> {
    let mut graph = flow_dsl::parse(dsl).map_err(|e| {
        format!(
            "DSL parse error at line {}:{} - {}",
            e.line, e.col, e.message
        )
    })?;
    dsl_semantics::repair_generated_graph(&mut graph);
    let issues = dsl_semantics::validate_generated_graph(&graph);
    if !issues.is_empty() {
        return Err(format!(
            "DSL semantic validation failed:\n{}",
            issues
                .iter()
                .map(|i| format!("- {i}"))
                .collect::<Vec<_>>()
                .join("\n")
        ));
    }
    Ok(flow_dsl::serialize(&graph))
}

/// Produce a targeted correction for parse errors the cloud model commonly
/// makes, injected into the corrective-retry prompt so the next attempt gets
/// concrete guidance rather than just the raw `line:col`. Returns `None` for
/// parse errors we have no specific remediation for.
fn parse_error_hint(message: &str) -> Option<String> {
    // Malformed string literal - almost always a multi-line value (e.g. file
    // `content`) or an unescaped inner double quote. Restate the string rules
    // concretely; the bare `unterminated string` line:col doesn't tell the
    // model *how* to fix it.
    if message.contains("unterminated string") {
        return Some(
            "IMPORTANT: a string literal is malformed. Flow DSL strings must be \
             single-line and double-quoted. Escape every inner double quote as \\\" \
             and write newlines as \\n - never put a real line break inside quotes. \
             For multi-line file content, keep it on one physical line using \\n, \
             e.g. content: \"line one\\nline two\\n\"."
                .to_string(),
        );
    }

    // Malformed edge: `a --> b` wants a node id immediately after `-->`, on one
    // line. Models trip this by breaking the edge across lines, quoting the
    // target, or leaving it empty.
    if message.contains("expected target identifier") || message.contains("after `-->`") {
        return Some(
            "IMPORTANT: an edge is malformed. Connect nodes by id on a single line: \
             `sourceId --> targetId` (chain as `a --> b --> c`). Put the target node id \
             immediately after `-->`; do not quote it, leave it empty, or split the edge \
             across lines."
                .to_string(),
        );
    }

    // ``unknown node type `fs` - supported: …``: an adapter name used as a node
    // type (a common mistake: emitting `shell[...]` / `fs[...]` as headers).
    // Produce a targeted correction showing the `action` + `adapter` form.
    let kind = message
        .split_once("unknown node type `")
        .and_then(|(_, rest)| rest.split('`').next())
        .filter(|k| !k.is_empty())?;
    let example = if matches!(kind, "cli" | "github" | "zowe") {
        "list[action: \"Run tool\"] {\n  adapter: \"cli\"\n  actionId: \"cli-tool\"\n  bin: \"<tool>\"\n  command: \"<subcommand>\"\n}"
    } else {
        "step[action: \"Run\"] {\n  adapter: \"ADAPTER\"\n  actionId: \"<see spec>\"\n}"
    }
    .replace("ADAPTER", kind);
    Some(format!(
        "IMPORTANT: `{kind}` is not a node type. The only node types are \
         `action`, `ai`, `agentic`, `utility`, `service`. `{kind}` is an adapter - \
         use an `action` node with `adapter: \"{kind}\"`, e.g.:\n\
         {example}\n\
         Do NOT write `{kind}[...]` as the header type."
    ))
}

#[cfg(test)]
mod parse_error_hint_tests {
    use super::parse_error_hint;

    #[test]
    fn hint_fires_for_unknown_node_type() {
        let msg = "unknown node type `fs` - supported: action, ai, agentic, utility, service";
        let hint = parse_error_hint(msg).expect("should produce a hint");
        assert!(hint.contains("`fs` is not a node type"));
        assert!(hint.contains("adapter: \"fs\""));
    }

    #[test]
    fn hint_fires_for_unterminated_string() {
        let hint = parse_error_hint("unterminated string").expect("should produce a hint");
        assert!(hint.contains("single-line"), "hint: {hint}");
        assert!(hint.contains("\\n"), "hint: {hint}");
        // The newline-before-quote variant carries the same substring.
        assert!(
            parse_error_hint("unterminated string (newline before closing quote)").is_some()
        );
    }

    #[test]
    fn hint_fires_for_malformed_edge() {
        let hint = parse_error_hint("expected target identifier after `-->`")
            .expect("should produce a hint");
        assert!(hint.contains("edge is malformed"), "hint: {hint}");
        assert!(hint.contains("-->"), "hint: {hint}");
    }

    #[test]
    fn no_hint_for_unrelated_parse_errors() {
        assert!(parse_error_hint("expected `}` but found end of input").is_none());
        assert!(parse_error_hint("unknown node type ``").is_none());
    }
}

/// End-to-end coverage of the cloud generation pipeline (parse → corrective
/// retry → recover) driven by a scripted stub provider, so no network or API
/// key is involved. Exercises the two robustness fixes: chained-arrow edges
/// must flow through generation without a retry, and an `unterminated string`
/// must trigger a corrective retry that carries the targeted string hint.
#[cfg(test)]
mod generation_e2e_tests {
    use super::*;
    use flow_adapter_ai::{CloudAiError, CloudAiProvider, CloudAiRequest, CloudAiResponse};
    use std::collections::VecDeque;
    use std::sync::Mutex;

    /// A `CloudAiProvider` that replays a fixed script of responses and records
    /// every user prompt it receives.
    struct ScriptedProvider {
        responses: Mutex<VecDeque<String>>,
        prompts: Mutex<Vec<String>>,
        finish_reason: String,
    }

    impl ScriptedProvider {
        fn new(responses: Vec<&str>) -> Arc<Self> {
            Self::new_with_finish(responses, "stop")
        }
        /// Replays the same `finish_reason` on every response - lets a test
        /// simulate a token-limit cut-off (`"length"` / `"max_tokens"`).
        fn new_with_finish(responses: Vec<&str>, finish_reason: &str) -> Arc<Self> {
            Arc::new(Self {
                responses: Mutex::new(responses.into_iter().map(String::from).collect()),
                prompts: Mutex::new(Vec::new()),
                finish_reason: finish_reason.to_string(),
            })
        }
    }

    #[async_trait::async_trait]
    impl CloudAiProvider for ScriptedProvider {
        fn name(&self) -> &str {
            "stub"
        }
        fn env_var(&self) -> &str {
            "STUB_KEY"
        }
        fn default_models(&self) -> &[&str] {
            &["stub-model"]
        }
        async fn invoke(&self, req: &CloudAiRequest) -> Result<CloudAiResponse, CloudAiError> {
            self.prompts.lock().unwrap().push(req.prompt.clone());
            let text = self
                .responses
                .lock()
                .unwrap()
                .pop_front()
                .expect("scripted provider ran out of responses");
            Ok(CloudAiResponse {
                provider: "stub".into(),
                model: req.model.clone(),
                text,
                finish_reason: self.finish_reason.clone(),
                input_tokens: 0,
                output_tokens: 0,
                latency_ms: 0,
            })
        }
    }

    fn app_with(stub: Arc<ScriptedProvider>) -> FlowApp {
        let app = FlowApp::with_store_and_settings(
            Store::open_in_memory().expect("in-memory store"),
            SettingsStore::in_memory(),
        )
        .with_replaced_cloud_providers(vec![stub])
        .with_in_memory_credentials();
        app.settings
            .update(SettingsPatch {
                allow_cloud_ai: Some(true),
                ..Default::default()
            })
            .expect("enable cloud ai");
        app.set_cloud_ai_key("stub", "k").expect("seed key");
        app
    }

    const CHAIN_DSL: &str = r#"flow "chain" v1

a[utility: "A"] {
  adapter: "utility"
  actionId: "log"
  message: "a"
  level: "info"
}

b[utility: "B"] {
  adapter: "utility"
  actionId: "log"
  message: "b"
  level: "info"
}

c[utility: "C"] {
  adapter: "utility"
  actionId: "log"
  message: "c"
  level: "info"
}

a --> b --> c
"#;

    #[tokio::test]
    async fn chained_arrows_from_model_parse_without_a_retry() {
        let stub = ScriptedProvider::new(vec![CHAIN_DSL]);
        let app = app_with(stub.clone());
        let dsl = app
            .generate_dsl_via_cloud("stub", "stub-model", "make a 3-step log chain", Some(512), None)
            .await
            .expect("chained-arrow DSL should generate without error");
        let graph = flow_dsl::parse(&dsl).expect("returned DSL parses");
        assert_eq!(graph.edges.len(), 2, "a --> b --> c desugars to two edges");
        assert_eq!(
            stub.prompts.lock().unwrap().len(),
            1,
            "valid-on-first-try DSL needs no corrective retry"
        );
    }

    #[tokio::test]
    async fn multiline_string_from_model_is_auto_sanitized_without_a_retry() {
        // The model emits a real line break inside a string (a multi-line file
        // `content` value) - a hard `unterminated string` parse error without
        // sanitizing. The pre-parse sanitizer escapes the break to `\n`, so
        // generation succeeds on the first try with no corrective retry.
        let multiline = "flow \"x\" v1\n\nwrite[action: \"Write\"] {\n  adapter: \"fs\"\n  actionId: \"write-file\"\n  workspaceRoot: \"/tmp/x\"\n  path: \"notes.txt\"\n  content: \"line one\nline two\n\"\n}\n";
        let stub = ScriptedProvider::new(vec![multiline]);
        let app = app_with(stub.clone());
        let dsl = app
            .generate_dsl_via_cloud("stub", "stub-model", "write a two-line file", Some(512), None)
            .await
            .expect("multi-line string should be sanitized and generate cleanly");
        let graph = flow_dsl::parse(&dsl).expect("sanitized DSL parses");
        let content = graph.nodes[0].data["content"].as_str().unwrap();
        assert!(
            content.contains('\n'),
            "the escaped \\n is unescaped back to a real newline in the value"
        );
        assert_eq!(
            stub.prompts.lock().unwrap().len(),
            1,
            "auto-sanitized DSL needs no corrective retry"
        );
    }

    // A node whose last field is missing its closing quote: the pre-parse
    // sanitizer can't rescue a genuinely unterminated string, so the lexer
    // reports `unterminated string` and a corrective retry must fire.
    const BAD_UNTERMINATED: &str = r#"flow "x" v1

n[utility: "N"] {
  adapter: "utility"
  actionId: "log"
  message: "hello"
  level: "info
}
"#;

    #[tokio::test]
    async fn unterminated_string_triggers_caret_retry_and_recovers() {
        let stub = ScriptedProvider::new(vec![BAD_UNTERMINATED, CHAIN_DSL]);
        let app = app_with(stub.clone());
        let dsl = app
            .generate_dsl_via_cloud("stub", "stub-model", "make a chain", Some(512), None)
            .await
            .expect("recovers on the corrective retry");
        flow_dsl::parse(&dsl).expect("recovered DSL parses");
        let prompts = stub.prompts.lock().unwrap();
        assert_eq!(prompts.len(), 2, "exactly one corrective retry");
        let retry = &prompts[1];
        assert!(
            retry.contains("The error is here:"),
            "retry shows the offending source line: {retry}"
        );
        assert!(retry.contains('^'), "retry carries a caret pointer: {retry}");
        assert!(
            retry.contains("single-line"),
            "retry restates the string rule: {retry}"
        );
    }

    #[tokio::test]
    async fn truncated_response_reports_token_limit_without_burning_retries() {
        // finish_reason "length" = cut off at max_tokens mid-`content`. Retrying
        // at the same budget only reproduces it, so generation must short-circuit
        // with a clear token-limit error after the first call.
        let stub = ScriptedProvider::new_with_finish(vec![BAD_UNTERMINATED], "length");
        let app = app_with(stub.clone());
        let err = app
            .generate_dsl_via_cloud("stub", "stub-model", "scaffold an app", Some(2048), None)
            .await
            .expect_err("a truncated response is an error");
        let msg = err.to_string();
        assert!(
            msg.contains("cut off at the token limit"),
            "clear truncation error, not 'unterminated string': {msg}"
        );
        assert_eq!(
            stub.prompts.lock().unwrap().len(),
            1,
            "truncation short-circuits - no wasted corrective retries"
        );
    }

    #[tokio::test]
    async fn exhausted_retries_name_the_provider_in_the_error() {
        // initial attempt + MAX_CORRECTIVE_RETRIES, all unterminated.
        let stub = ScriptedProvider::new(vec![
            BAD_UNTERMINATED,
            BAD_UNTERMINATED,
            BAD_UNTERMINATED,
            BAD_UNTERMINATED,
        ]);
        let app = app_with(stub.clone());
        let err = app
            .generate_dsl_via_cloud("stub", "stub-model", "make a chain", Some(512), None)
            .await
            .expect_err("every attempt fails to parse");
        let msg = err.to_string();
        assert!(
            msg.contains("stub model returned invalid DSL"),
            "error names the provider, not a hardcoded 'cloud': {msg}"
        );
        assert!(
            msg.contains("3 corrective retries"),
            "error names the retry budget: {msg}"
        );
    }
}

#[cfg(test)]
mod escape_newlines_tests {
    use super::escape_newlines_in_strings;

    #[test]
    fn escapes_newline_inside_a_string() {
        assert_eq!(
            escape_newlines_in_strings("content: \"a\nb\""),
            "content: \"a\\nb\""
        );
    }

    #[test]
    fn leaves_structural_newlines_outside_strings() {
        let src = "a: \"x\"\nb: \"y\"\n";
        assert_eq!(escape_newlines_in_strings(src), src);
    }

    #[test]
    fn honors_escaped_quote_so_string_stays_open() {
        // The `\"` does not close the string, so the following newline is still
        // inside it and gets escaped.
        assert_eq!(
            escape_newlines_in_strings("\"a\\\"b\nc\""),
            "\"a\\\"b\\nc\""
        );
    }

    #[test]
    fn quotes_inside_comments_are_literal() {
        // The `"` in the comment must not open a string, so the newline ending
        // the comment stays a real newline.
        let src = "# a \"quote\n\"real\nx\"";
        assert_eq!(escape_newlines_in_strings(src), "# a \"quote\n\"real\\nx\"");
    }
}

#[cfg(test)]
mod existing_binary_tests {
    use super::FlowApp;

    #[test]
    fn drops_missing_and_empty_paths() {
        assert_eq!(FlowApp::existing_binary(None), None);
        assert_eq!(FlowApp::existing_binary(Some("   ".into())), None);
        // A stale build path that no longer exists must not be returned, so
        // resolution can fall through to a managed engine / $PATH.
        assert_eq!(
            FlowApp::existing_binary(Some(
                "/nonexistent/target/debug/resources/llama/llama-server".into()
            )),
            None
        );
    }

    #[test]
    fn keeps_an_existing_file() {
        // The test binary itself is a real, executable file on disk.
        let me = std::env::current_exe().unwrap();
        let path = me.to_string_lossy().to_string();
        assert_eq!(FlowApp::existing_binary(Some(path.clone())), Some(path));
    }
}

impl FlowApp {
    pub fn new() -> Self {
        let store = Self::open_default_store().unwrap_or_else(|e| {
            tracing::warn!(error = %e, "failed to open persistent store; falling back to in-memory");
            Store::open_in_memory().expect("in-memory store init")
        });
        let settings = SettingsStore::open(Self::settings_path());
        Self::with_store_and_settings(store, settings)
    }

    /// Test-only: replace the cloud-provider list. Lets unit tests inject
    /// a stub `CloudAiProvider` so `generate_dsl_via_cloud` and
    /// `generate_dsl_auto` can be exercised without a real network call.
    /// Out of the public surface - gated to `#[cfg(test)]` so a future
    /// caller doesn't reach in and quietly blank out the live providers.
    #[cfg(test)]
    pub(crate) fn with_replaced_cloud_providers(
        mut self,
        providers: Vec<Arc<dyn CloudAiProvider>>,
    ) -> Self {
        self.cloud_providers = providers;
        self
    }

    /// Test-only: swap the OS-keyring-backed credential store for an
    /// in-memory one. The default `KeyringCredentialStore` prompts the
    /// user via macOS Keychain Access on the **first** read of a
    /// previously-unseen service namespace - and a cancelled prompt
    /// surfaces as a `Keyring(...)` error, *not* `NotFound`, so the
    /// env-var fallback in `EnvFallbackResolver` never fires. Tests
    /// that need to exercise the cloud-AI path must therefore avoid
    /// the keyring entirely.
    #[cfg(test)]
    pub(crate) fn with_in_memory_credentials(mut self) -> Self {
        use flow_security::InMemoryCredentialStore;

        let store: Arc<dyn CredentialStore> = Arc::new(InMemoryCredentialStore::new());
        self.resolver = Arc::new(EnvFallbackResolver::new(store.clone()));
        self.credentials = store;
        self
    }

    pub fn with_store_and_settings(store: Store, settings: SettingsStore) -> Self {
        // Persist the sidecar credential index alongside the app's other
        // on-disk state. The index records *which* accounts have keychain
        // entries (names only, no secrets) so `exists()` answers without
        // a keychain prompt - see KeyringCredentialStore for the rationale.
        let index_path = Self::flow_dir().join("credential_index.json");
        let credentials: Arc<dyn CredentialStore> =
            Arc::new(KeyringCredentialStore::with_index(index_path));
        let resolver: Arc<dyn CredentialResolver> =
            Arc::new(EnvFallbackResolver::new(credentials.clone()));

        let connections = ConnectionStore::open(&Self::flow_dir(), credentials.clone())
            .unwrap_or_else(|e| {
                tracing::warn!(error = %e, "failed to open connections store; using in-memory");
                ConnectionStore::open(&std::env::temp_dir(), credentials.clone())
                    .expect("in-memory connections init")
            });

        let sanitizer = Arc::new(PiiSanitizer::new());
        let mut adapters = AdapterRegistry::new();
        adapters.register(Arc::new(MockAdapter::new("mock")));
        // One vendor-neutral CLI adapter drives every cli-tool node; the binary
        // to run comes from each node's `bin` field, not a per-vendor adapter.
        adapters.register(Arc::new(CliAdapter::new()));
        adapters.register(Arc::new(MockAdapter::new("zosmf")));
        adapters.register(Arc::new(MockAdapter::new("ssh")));
        adapters.register(Arc::new(ShellAdapter::new()));
        adapters.register(Arc::new(FsAdapter::new()));
        adapters.register(Arc::new(UtilityAdapter::new(Self::downloads_dir())));
        // Generic, data-driven `service` adapter: every external API integration
        // is catalog data (each entry's `serviceIntegration`), so this one adapter
        // backs all service nodes. Credentials/tokens come from the keyring.
        adapters.register(Arc::new(flow_adapter_service::ServiceAdapter::new(
            Arc::new(nodes::service_integrations()),
            credentials.clone(),
        )));

        let cloud_providers: Vec<Arc<dyn CloudAiProvider>> = vec![
            Arc::new(ClaudeProvider::new()),
            Arc::new(OpenAiProvider::new()),
            Arc::new(GeminiProvider::new()),
            Arc::new(NvidiaProvider::new()),
            Arc::new(DeepSeekProvider::new()),
            Arc::new(LocalOpenAiProvider::new()),
        ];

        // Seed the system-managed Default template collection row. Failure is
        // non-fatal: with no row, list/save still treats "no membership" as
        // Default, and the row will be re-seeded on the next successful boot.
        if let Err(e) = templates::ensure_default(&store) {
            tracing::warn!(error = ?e, "failed to ensure default template collection");
        }

        // Restore durable working memory so `{{memory.<key>}}` survives restarts
        // (roadmap E1 durable agent memory). Best-effort: a load failure just
        // starts empty.
        let initial_memory = store.load_memory().unwrap_or_default();

        Self {
            sanitizer,
            adapters: Arc::new(adapters),
            cloud_providers,
            credentials,
            resolver,
            connections,
            settings,
            store,
            llm_server: crate::llm_server::LlmServerHandle::new(),
            stream_sink: std::sync::RwLock::new(Arc::new(flow_adapter_ai::NullStreamSink)),
            working_memory: Arc::new(std::sync::Mutex::new(initial_memory)),
            runs: Default::default(),
            download_cancel: Arc::new(std::sync::atomic::AtomicBool::new(false)),
            confirm_gate_disabled: std::sync::atomic::AtomicBool::new(false),
            workspace_base: WorkspaceBase::default(),
        }
    }

    /// Override the edition default workspace. flow-cli calls this with
    /// `Fixed(<launch dir or --workspace>)`; desktop/server keep `Scratch`.
    pub fn with_workspace_base(mut self, base: WorkspaceBase) -> Self {
        self.workspace_base = base;
        self
    }

    /// Resolve the execution workspace for a flow: the stored per-flow override
    /// (if any) else this host's edition default. The path is created before a
    /// run so fs nodes (which require an existing root) work against it.
    fn resolve_workspace(&self, flow_id: &str) -> PathBuf {
        self.store
            .get_flow_workspace(flow_id)
            .ok()
            .flatten()
            .filter(|s| !s.trim().is_empty())
            .map(|s| expand_home(&s))
            .unwrap_or_else(|| self.workspace_base.resolve(flow_id))
    }

    /// The per-flow workspace override, or `None` when the flow uses the default.
    pub fn flow_workspace(&self, flow_id: &str) -> Option<String> {
        self.store.get_flow_workspace(flow_id).ok().flatten()
    }

    /// The default workspace this host would use for `flow_id` absent an
    /// override - surfaced to the UI so it can show the resolved path.
    pub fn default_workspace(&self, flow_id: &str) -> String {
        self.workspace_base.resolve(flow_id).to_string_lossy().into_owned()
    }

    pub fn set_flow_workspace(&self, flow_id: &str, path: &str) -> Result<(), AppError> {
        self.store
            .set_flow_workspace(flow_id, path)
            .map_err(|e| AppError::Storage(e.to_string()))
    }

    pub fn clear_flow_workspace(&self, flow_id: &str) -> Result<(), AppError> {
        self.store
            .clear_flow_workspace(flow_id)
            .map_err(|e| AppError::Storage(e.to_string()))
    }

    /// Install the LLM token sink for streaming events. The host (Tauri) wires
    /// a sink that re-emits each delta as a `flow:llm_token` Tauri event so
    /// the header chip can render the rolling tail; headless callers can keep
    /// the default `NullStreamSink`. Safe to call at any time from any thread.
    pub fn set_stream_sink(&self, sink: Arc<dyn flow_adapter_ai::LlmStreamSink>) {
        *self.stream_sink.write().expect("stream_sink write") = sink;
    }

    /// Snapshot the current sink so callers can hold a reference for the
    /// duration of an LLM call without blocking concurrent sink swaps.
    pub fn stream_sink(&self) -> Arc<dyn flow_adapter_ai::LlmStreamSink> {
        self.stream_sink.read().expect("stream_sink read").clone()
    }

    pub fn list_connections(&self) -> Vec<flow_domain::connection::ConnectionProfile> {
        self.connections.list()
    }

    pub fn save_connection(
        &self,
        profile: flow_domain::connection::ConnectionProfile,
        secret: Option<&str>,
    ) -> Result<(), AppError> {
        self.connections
            .upsert(profile, secret)
            .map_err(|e| AppError::Credential(e.to_string()))
    }

    pub fn delete_connection(&self, id: &str) -> Result<(), AppError> {
        self.connections
            .delete(id)
            .map_err(|e| AppError::Credential(e.to_string()))
    }

    pub fn connection_has_secret(&self, id: &str) -> bool {
        self.connections.has_secret(id)
    }

    pub fn resolve_zowe_connection(&self, id: &str) -> Result<ResolvedZoweConnection, AppError> {
        self.connections
            .resolve_zowe(id)
            .map_err(|e| AppError::Credential(e.to_string()))
    }

    /// True when `provider` matches a registered cloud-AI provider name.
    fn is_known_cloud_provider(&self, provider: &str) -> bool {
        self.cloud_providers.iter().any(|p| p.name() == provider)
    }

    pub fn set_cloud_ai_key(&self, provider: &str, secret: &str) -> Result<(), AppError> {
        if !self.is_known_cloud_provider(provider) {
            return Err(AppError::Credential(format!(
                "unknown cloud provider '{provider}'"
            )));
        }
        let account = CredentialKind::CloudAiKey.account_for(provider);
        self.credentials
            .set(&account, secret)
            .map_err(|e| AppError::Credential(e.to_string()))
    }

    pub fn delete_cloud_ai_key(&self, provider: &str) -> Result<(), AppError> {
        if !self.is_known_cloud_provider(provider) {
            return Err(AppError::Credential(format!(
                "unknown cloud provider '{provider}'"
            )));
        }
        let account = CredentialKind::CloudAiKey.account_for(provider);
        self.credentials
            .delete(&account)
            .map_err(|e| AppError::Credential(e.to_string()))
    }

    pub fn cloud_ai_key_exists(&self, provider: &str) -> bool {
        if !self.is_known_cloud_provider(provider) {
            return false;
        }
        let account = CredentialKind::CloudAiKey.account_for(provider);
        self.credentials.exists(&account)
    }

    // ---- Service-node connections ----
    //
    // A `service` node's connection secret lives in the keyring under
    // `service:<slug>`: a raw API key / bearer token / `user:pass` for basic
    // auth, or (for OAuth2) the JSON token bundle written by `service_oauth_*`.
    // The OAuth client id/secret (operator-supplied) live under
    // `service:<slug>:client_id` / `:client_secret`.

    fn service_kv_account(slug: &str, suffix: &str) -> String {
        CredentialKind::Service.account_for(&format!("{slug}:{suffix}"))
    }

    fn service_integration_for(slug: &str) -> Result<crate::nodes::ServiceIntegration, AppError> {
        crate::nodes::entry_by_slug(slug)
            .and_then(|e| e.service_integration)
            .ok_or_else(|| AppError::Credential(format!("`{slug}` is not a service node")))
    }

    /// Store a raw connection secret (api-key / bearer token / `user:pass`) for a
    /// service node, keyed by its catalog slug.
    pub fn set_service_secret(&self, slug: &str, secret: &str) -> Result<(), AppError> {
        let account = CredentialKind::Service.account_for(slug);
        self.credentials
            .set(&account, secret)
            .map_err(|e| AppError::Credential(e.to_string()))
    }

    /// Forget a service connection (token/secret + any OAuth client credentials).
    pub fn disconnect_service(&self, slug: &str) -> Result<(), AppError> {
        let _ = self.credentials.delete(&CredentialKind::Service.account_for(slug));
        let _ = self.credentials.delete(&Self::service_kv_account(slug, "client_id"));
        let _ = self.credentials.delete(&Self::service_kv_account(slug, "client_secret"));
        Ok(())
    }

    /// True when a connection secret/token is stored for the service.
    pub fn service_connection_exists(&self, slug: &str) -> bool {
        self.credentials
            .exists(&CredentialKind::Service.account_for(slug))
    }

    /// Persist the operator-supplied OAuth2 client id + secret for a service.
    pub fn set_service_oauth_client(
        &self,
        slug: &str,
        client_id: &str,
        client_secret: &str,
    ) -> Result<(), AppError> {
        self.credentials
            .set(&Self::service_kv_account(slug, "client_id"), client_id)
            .map_err(|e| AppError::Credential(e.to_string()))?;
        self.credentials
            .set(&Self::service_kv_account(slug, "client_secret"), client_secret)
            .map_err(|e| AppError::Credential(e.to_string()))
    }

    /// Build the OAuth2 consent URL the user opens to authorize a service.
    /// `redirect_uri` is the loopback the desktop app captures the code on.
    pub fn service_oauth_authorize_url(
        &self,
        slug: &str,
        redirect_uri: &str,
        state: &str,
    ) -> Result<String, AppError> {
        let integ = Self::service_integration_for(slug)?;
        if integ.auth.scheme != "oauth2" {
            return Err(AppError::Credential(format!("`{slug}` does not use OAuth2")));
        }
        let auth_url = integ
            .auth
            .auth_url
            .as_deref()
            .ok_or_else(|| AppError::Credential(format!("`{slug}` integration has no authUrl")))?;
        let client_id = self
            .credentials
            .get(&Self::service_kv_account(slug, "client_id"))
            .map_err(|_| {
                AppError::Credential(format!("set the OAuth client id for `{slug}` first"))
            })?;
        Ok(flow_security::oauth::build_authorize_url(
            auth_url,
            &client_id,
            redirect_uri,
            &integ.auth.scopes,
            state,
        ))
    }

    /// Exchange an OAuth2 authorization code and persist the token bundle.
    pub async fn service_oauth_complete(
        &self,
        slug: &str,
        code: &str,
        redirect_uri: &str,
    ) -> Result<(), AppError> {
        let integ = Self::service_integration_for(slug)?;
        let token_url = integ
            .auth
            .token_url
            .as_deref()
            .ok_or_else(|| AppError::Credential(format!("`{slug}` integration has no tokenUrl")))?;
        let client_id = self
            .credentials
            .get(&Self::service_kv_account(slug, "client_id"))
            .map_err(|_| AppError::Credential(format!("missing OAuth client id for `{slug}`")))?;
        let client_secret = self
            .credentials
            .get(&Self::service_kv_account(slug, "client_secret"))
            .map_err(|_| AppError::Credential(format!("missing OAuth client secret for `{slug}`")))?;
        let token = flow_security::oauth::exchange_code(
            token_url,
            &client_id,
            &client_secret,
            code,
            redirect_uri,
        )
        .await
        .map_err(|e| AppError::Credential(e.to_string()))?;
        flow_security::oauth::store_token(self.credentials.as_ref(), slug, &token)
            .map_err(|e| AppError::Credential(e.to_string()))
    }

    fn home_dir() -> PathBuf {
        PathBuf::from(
            std::env::var("HOME")
                .or_else(|_| std::env::var("USERPROFILE"))
                .unwrap_or_else(|_| ".".into()),
        )
    }

    /// The app's data directory (db, settings, installed nodes/templates,
    /// models, downloads, connections). Honors the `FLOW_STUDIO_DIR` env
    /// override - the `desktop:dev:fresh` dev script points it at a throwaway
    /// directory so every run starts from a fresh profile. Falls back to
    /// `<home>/.flow-studio` when unset.
    fn flow_dir() -> PathBuf {
        if let Some(dir) = std::env::var_os("FLOW_STUDIO_DIR") {
            if !dir.is_empty() {
                return PathBuf::from(dir);
            }
        }
        Self::home_dir().join(".flow-studio")
    }

    fn open_default_store() -> Result<Store, flow_storage::StoreError> {
        let mut path = Self::flow_dir();
        path.push("db");
        let _ = std::fs::create_dir_all(&path);
        path.push("flow.sqlite");
        Store::open(path)
    }

    fn templates_dir() -> PathBuf {
        let mut path = Self::flow_dir();
        path.push("templates");
        let _ = std::fs::create_dir_all(&path);
        path
    }

    /// Where downloaded LLMs live: `<flow_dir()>/llms/`. The managed
    /// `llama-server` loads models from here.
    pub fn llms_dir() -> PathBuf {
        let mut path = Self::flow_dir();
        path.push("llms");
        let _ = std::fs::create_dir_all(&path);
        path
    }

    /// Where the managed model-server engine is fetched on first use:
    /// `<flow_dir()>/engines/`. Shared by every edition on this host.
    pub fn engines_dir() -> PathBuf {
        let mut path = Self::flow_dir();
        path.push("engines");
        let _ = std::fs::create_dir_all(&path);
        path
    }

    /// Where the `utility:download` action writes artifacts:
    /// `<flow_dir()>/downloads/`. Created lazily on first access.
    pub fn downloads_dir() -> PathBuf {
        let mut path = Self::flow_dir();
        path.push("downloads");
        let _ = std::fs::create_dir_all(&path);
        path
    }

    /// Where third-party node schemes installed via the Node Hub live:
    /// `<flow_dir()>/nodes/<slug>.json`. Created lazily.
    pub fn nodes_dir() -> PathBuf {
        let mut path = Self::flow_dir();
        path.push("nodes");
        let _ = std::fs::create_dir_all(&path);
        path
    }

    /// The Model Hub catalog (the static registry loaded from `hub_catalog.json`).
    pub fn hub_catalog(&self) -> Vec<crate::hub::HubModel> {
        crate::hub::catalog()
    }

    /// Probe this machine for the device-compatibility check: OS/arch, total +
    /// available RAM, and free disk on the volume holding the LLMs directory.
    pub fn system_info(&self) -> SystemInfo {
        use sysinfo::{Disks, System};
        const GIB: f64 = 1024.0 * 1024.0 * 1024.0;
        let round1 = |v: f64| (v * 10.0).round() / 10.0;

        let mut sys = System::new();
        sys.refresh_memory();
        let total = sys.total_memory() as f64 / GIB;
        let avail = sys.available_memory() as f64 / GIB;

        // Free space on the disk whose mount point is the longest prefix of the
        // LLMs download dir.
        let llms = Self::llms_dir();
        let disks = Disks::new_with_refreshed_list();
        let mut free_disk: u64 = 0;
        let mut best_len = 0usize;
        for d in disks.list() {
            let mp = d.mount_point();
            if llms.starts_with(mp) && mp.as_os_str().len() >= best_len {
                best_len = mp.as_os_str().len();
                free_disk = d.available_space();
            }
        }

        let cpu_count = std::thread::available_parallelism()
            .map(|n| n.get() as u32)
            .unwrap_or(0);

        SystemInfo {
            os: std::env::consts::OS.to_string(),
            arch: std::env::consts::ARCH.to_string(),
            total_ram_gb: round1(total),
            available_ram_gb: round1(avail),
            free_disk_gb: round1(free_disk as f64 / GIB),
            cpu_count,
        }
    }

    /// LLMs already downloaded into the local LLMs directory.
    pub fn list_local_llms(&self) -> Vec<crate::hub::LocalLlm> {
        crate::hub::list_local_llms(&Self::llms_dir())
    }

    /// Installed models: downloaded LLMs matched to a catalog model whose
    /// download-option URL produces that file name.
    pub async fn hub_installed(&self) -> Vec<crate::hub::InstalledModel> {
        use crate::hub::{HubFormat, InstalledModel};
        let catalog = crate::hub::catalog();
        let mut out: Vec<InstalledModel> = Vec::new();

        // LLMs: match a downloaded file name to a catalog model whose option
        // URL produces that file name.
        for llm in self.list_local_llms() {
            if let Some(m) = catalog.iter().find(|m| {
                m.download_options
                    .iter()
                    .any(|o| o.url.rsplit('/').next() == Some(llm.file_name.as_str()))
            }) {
                if out.iter().any(|i| i.id == m.id) {
                    continue;
                }
                out.push(InstalledModel {
                    id: m.id.clone(),
                    version: m.latest_version.clone(),
                    format: HubFormat::Gguf,
                    update_available: false,
                });
            }
        }
        out
    }

    /// Download a catalog model's variant into the local LLMs directory,
    /// reporting progress through `cb`. Returns the on-disk path.
    pub async fn download_hub_model(
        &self,
        id: &str,
        variant: Option<&str>,
        cb: Option<crate::install::ProgressCallback>,
    ) -> Result<PathBuf, AppError> {
        let model = crate::hub::find(id)
            .ok_or_else(|| AppError::AiInvocation(format!("unknown hub model '{id}'")))?;
        // Fresh download: clear any leftover cancel request from a prior one.
        self.download_cancel
            .store(false, std::sync::atomic::Ordering::Relaxed);
        crate::hub::download(
            &model,
            variant,
            &Self::llms_dir(),
            cb,
            Some(self.download_cancel.clone()),
        )
        .await
        .map_err(AppError::AiInvocation)
    }

    /// Request cancellation of the in-flight Hub download (if any). The
    /// streaming loop in `hub::download` notices on its next chunk, removes the
    /// partial file, and returns a "download cancelled" error.
    pub fn cancel_download(&self) {
        self.download_cancel
            .store(true, std::sync::atomic::Ordering::Relaxed);
    }

    /// Delete a downloaded LLM from the local LLMs directory (confined to it).
    pub fn delete_local_llm(&self, file_name: &str) -> Result<(), AppError> {
        crate::hub::delete_local_llm(&Self::llms_dir(), file_name)
            .map_err(AppError::AiInvocation)
    }

    pub fn list_templates(&self) -> Result<Vec<TemplateRecord>, AppError> {
        templates::list(&Self::templates_dir(), &self.store)
            .map_err(|e| AppError::Template(e.to_string()))
    }

    pub fn list_templates_in(
        &self,
        collection_slug: &str,
    ) -> Result<Vec<TemplateRecord>, AppError> {
        templates::list_in(&Self::templates_dir(), &self.store, collection_slug)
            .map_err(|e| AppError::Template(e.to_string()))
    }

    pub fn save_template(&self, name: &str, graph: FlowGraph) -> Result<TemplateRecord, AppError> {
        templates::save(
            &Self::templates_dir(),
            &self.store,
            name,
            &graph,
            templates::DEFAULT_COLLECTION_SLUG,
            None,
        )
        .map_err(|e| AppError::Template(e.to_string()))
    }

    pub fn save_template_in(
        &self,
        name: &str,
        graph: FlowGraph,
        collection_slug: &str,
        source: Option<templates::TemplateSource>,
    ) -> Result<TemplateRecord, AppError> {
        templates::save(
            &Self::templates_dir(),
            &self.store,
            name,
            &graph,
            collection_slug,
            source,
        )
        .map_err(|e| AppError::Template(e.to_string()))
    }

    pub fn load_template(&self, slug: &str) -> Result<FlowGraph, AppError> {
        templates::load(&Self::templates_dir(), slug).map_err(|e| AppError::Template(e.to_string()))
    }

    pub fn delete_template(&self, slug: &str) -> Result<(), AppError> {
        templates::delete(&Self::templates_dir(), &self.store, slug)
            .map_err(|e| AppError::Template(e.to_string()))?;
        // Drop any recurring schedule so the scheduler doesn't keep trying to
        // load a template that no longer exists (E11).
        let _ = self.store.delete_schedule(slug);
        Ok(())
    }

    /// Every persisted flow schedule (roadmap E11), newest-updated first.
    pub fn list_schedules(&self) -> Result<Vec<ScheduleRecord>, AppError> {
        self.store
            .list_schedules()
            .map_err(|e| AppError::Storage(e.to_string()))
    }

    /// The schedule for a saved flow, if one exists.
    pub fn get_schedule(&self, template_slug: &str) -> Result<Option<ScheduleRecord>, AppError> {
        self.store
            .get_schedule(template_slug)
            .map_err(|e| AppError::Storage(e.to_string()))
    }

    /// Create or update a flow's schedule. Validates the frequency, computes the
    /// next fire time from the anchor (when enabled), and preserves `created_at`
    /// / `last_run_at` across updates.
    pub fn upsert_schedule(
        &self,
        patch: scheduler::SchedulePatch,
    ) -> Result<ScheduleRecord, AppError> {
        let freq = scheduler::ScheduleFrequency::parse(&patch.frequency).ok_or_else(|| {
            AppError::Validation(format!("unknown schedule frequency `{}`", patch.frequency))
        })?;
        let catchup = patch
            .catchup
            .as_deref()
            .and_then(scheduler::CatchUpPolicy::parse)
            .unwrap_or_default();
        let now = Utc::now();
        let existing = self
            .store
            .get_schedule(&patch.template_slug)
            .map_err(|e| AppError::Storage(e.to_string()))?;
        if patch.enabled {
            let graph = self.load_template(&patch.template_slug)?;
            if graph.nodes.is_empty() {
                return Err(AppError::Validation(
                    "cannot schedule an empty flow".into(),
                ));
            }
        }
        let timezone = patch
            .timezone
            .clone()
            .or_else(|| existing.as_ref().map(|e| e.timezone.clone()))
            .unwrap_or_else(|| "UTC".into());
        let options = scheduler::ScheduleOptions {
            timezone: Some(timezone.clone()),
            cron: patch.cron.clone(),
            every_minutes: patch.every_minutes,
            until: patch.until,
            max_runs: patch.max_runs,
            run_count: existing.as_ref().map(|e| e.run_count).unwrap_or(0),
        };
        let next_run_at = if patch.enabled {
            scheduler::next_run_after(freq, patch.anchor_at, now, &options)
                .map_err(AppError::Validation)?
        } else {
            None
        };
        let rec = ScheduleRecord {
            template_slug: patch.template_slug,
            collection_slug: patch.collection_slug,
            flow_name: patch.flow_name,
            enabled: patch.enabled,
            frequency: freq.as_str().to_string(),
            anchor_at: patch.anchor_at,
            timezone,
            cron: patch.cron,
            every_minutes: patch.every_minutes,
            until: patch.until,
            max_runs: patch.max_runs,
            catchup: catchup.as_str().to_string(),
            next_run_at,
            last_run_at: existing.as_ref().and_then(|e| e.last_run_at),
            last_status: existing.as_ref().and_then(|e| e.last_status.clone()),
            run_count: existing.as_ref().map(|e| e.run_count).unwrap_or(0),
            created_at: existing.as_ref().map(|e| e.created_at).unwrap_or(now),
            updated_at: now,
        };
        self.store
            .upsert_schedule(&rec)
            .map_err(|e| AppError::Storage(e.to_string()))?;
        Ok(rec)
    }

    pub fn preview_schedule(
        &self,
        req: scheduler::SchedulePreviewRequest,
    ) -> Result<Vec<DateTime<Utc>>, AppError> {
        scheduler::preview(req).map_err(AppError::Validation)
    }

    pub fn delete_schedule(&self, template_slug: &str) -> Result<(), AppError> {
        self.store
            .delete_schedule(template_slug)
            .map_err(|e| AppError::Storage(e.to_string()))
    }

    pub fn set_schedule_enabled(
        &self,
        template_slug: &str,
        enabled: bool,
    ) -> Result<Option<ScheduleRecord>, AppError> {
        let Some(schedule) = self.get_schedule(template_slug)? else {
            return Ok(None);
        };
        let next_run_at = if enabled {
            self.compute_next_schedule_run(&schedule, Utc::now(), schedule.run_count)?
        } else {
            None
        };
        self.store
            .set_schedule_enabled(template_slug, enabled, next_run_at)
            .map_err(|e| AppError::Storage(e.to_string()))?;
        self.get_schedule(template_slug)
    }

    pub fn migrate_schedule(
        &self,
        old_slug: &str,
        new_slug: &str,
        collection_slug: &str,
        flow_name: &str,
    ) -> Result<(), AppError> {
        if old_slug == new_slug {
            return Ok(());
        }
        self.store
            .migrate_schedule(old_slug, new_slug, collection_slug, flow_name)
            .map_err(|e| AppError::Storage(e.to_string()))
    }

    /// Enabled schedules due to fire now - the background scheduler's per-tick
    /// work list.
    pub fn due_schedules(&self) -> Result<Vec<ScheduleRecord>, AppError> {
        if !self.settings.scheduler_enabled() {
            return Ok(Vec::new());
        }
        self.store
            .list_due_schedules(Utc::now())
            .map_err(|e| AppError::Storage(e.to_string()))
    }

    /// Fire a scheduled flow: advance its timer first (so an overlapping tick
    /// won't double-fire), load the saved template, and run it under a fresh
    /// execution id with the run persisted to history (metadata only - the
    /// zero-egress boundary is preserved). Returns the run summary.
    pub async fn run_scheduled(
        &self,
        schedule: &ScheduleRecord,
    ) -> Result<Option<ExecutionSummary>, AppError> {
        let now = Utc::now();
        let scheduled_at = schedule.next_run_at.unwrap_or(now);
        let poll_grace = chrono::Duration::seconds((self.settings.scheduler_poll_secs() * 2) as i64);
        let missed = now - scheduled_at > poll_grace;
        let catchup = scheduler::CatchUpPolicy::parse(&schedule.catchup).unwrap_or_default();
        if missed && catchup == scheduler::CatchUpPolicy::Skip {
            let next = self.compute_next_schedule_run(schedule, now, schedule.run_count)?;
            let _ = self
                .store
                .advance_schedule_without_run(&schedule.template_slug, now, next, "skipped");
            return Ok(None);
        }

        let after = if missed && catchup == scheduler::CatchUpPolicy::RunAll {
            scheduled_at
        } else {
            now
        };
        let next = self.compute_next_schedule_run(
            schedule,
            after,
            schedule.run_count.saturating_add(1),
        )?;
        let _ = self
            .store
            .mark_schedule_run(&schedule.template_slug, now, next, "running");
        let graph = self.load_template(&schedule.template_slug)?;
        let events: Arc<dyn EventSink> =
            Arc::new(flow_execution::StorageSink::with_flow_name_and_trigger(
                self.store.clone(),
                schedule.flow_name.clone(),
                "scheduled".into(),
            ));
        match self
            .run_graph_internal(Uuid::new_v4().to_string(), graph, events, Some(false), None, None)
            .await
        {
            Ok(summary) => {
                let _ = self.store.advance_schedule_without_run(
                    &schedule.template_slug,
                    Utc::now(),
                    next,
                    &summary.status,
                );
                Ok(Some(summary))
            }
            Err(e) => {
                let _ = self.store.advance_schedule_without_run(
                    &schedule.template_slug,
                    Utc::now(),
                    next,
                    "failed",
                );
                Err(e)
            }
        }
    }

    /// Fire one due schedule, honoring its catch-up policy. `run-all` may
    /// produce multiple summaries when several fire times were missed while the
    /// process was down; other policies produce zero or one.
    pub async fn run_due_schedule(
        &self,
        schedule: &ScheduleRecord,
    ) -> Result<Vec<ExecutionSummary>, AppError> {
        let mut out = Vec::new();
        let mut current = schedule.clone();
        let run_all =
            scheduler::CatchUpPolicy::parse(&current.catchup) == Some(scheduler::CatchUpPolicy::RunAll);
        for _ in 0..100 {
            if let Some(summary) = self.run_scheduled(&current).await? {
                out.push(summary);
            }
            if !run_all {
                break;
            }
            let Some(next) = self.get_schedule(&current.template_slug)? else {
                break;
            };
            if !next.enabled || next.next_run_at.is_none_or(|n| n > Utc::now()) {
                break;
            }
            current = next;
        }
        Ok(out)
    }

    fn compute_next_schedule_run(
        &self,
        schedule: &ScheduleRecord,
        after: DateTime<Utc>,
        run_count: u32,
    ) -> Result<Option<DateTime<Utc>>, AppError> {
        let freq = scheduler::ScheduleFrequency::parse(&schedule.frequency).ok_or_else(|| {
            AppError::Validation(format!("unknown schedule frequency `{}`", schedule.frequency))
        })?;
        scheduler::next_run_after(
            freq,
            schedule.anchor_at,
            after,
            &scheduler::ScheduleOptions {
                timezone: Some(schedule.timezone.clone()),
                cron: schedule.cron.clone(),
                every_minutes: schedule.every_minutes,
                until: schedule.until,
                max_runs: schedule.max_runs,
                run_count,
            },
        )
        .map_err(AppError::Validation)
    }

    pub fn list_template_collections(
        &self,
    ) -> Result<Vec<flow_storage::TemplateCollectionRecord>, AppError> {
        self.store
            .list_template_collections()
            .map_err(|e| AppError::Template(e.to_string()))
    }

    /// Insert a new user collection from a display name. Slug is derived
    /// server-side via the same `slugify` rule as templates. Refuses on
    /// empty-after-slugify and on slug collision (idempotent re-creation
    /// callers should branch on the returned `AppError::Template` text).
    pub fn create_template_collection(
        &self,
        name: &str,
    ) -> Result<flow_storage::TemplateCollectionRecord, AppError> {
        let slug = templates::slugify(name);
        if slug.is_empty() {
            return Err(AppError::Template(
                "collection name is empty after slugification".into(),
            ));
        }
        if self
            .store
            .get_template_collection(&slug)
            .map_err(|e| AppError::Template(e.to_string()))?
            .is_some()
        {
            return Err(AppError::Template(format!(
                "collection `{slug}` already exists"
            )));
        }
        self.store
            .upsert_template_collection(&slug, name, false)
            .map_err(|e| AppError::Template(e.to_string()))
    }

    /// Idempotent get-or-create: returns the existing collection if a row
    /// already exists for the slugified name, otherwise inserts a new one.
    /// Used by the Ansible-collection importer (step 6) so re-importing the
    /// same tarball reuses the same collection row instead of failing on
    /// the slug-collision check in [`Self::create_template_collection`].
    pub fn ensure_template_collection(
        &self,
        name: &str,
    ) -> Result<flow_storage::TemplateCollectionRecord, AppError> {
        let slug = templates::slugify(name);
        if slug.is_empty() {
            return Err(AppError::Template(
                "collection name is empty after slugification".into(),
            ));
        }
        if let Some(existing) = self
            .store
            .get_template_collection(&slug)
            .map_err(|e| AppError::Template(e.to_string()))?
        {
            return Ok(existing);
        }
        self.store
            .upsert_template_collection(&slug, name, false)
            .map_err(|e| AppError::Template(e.to_string()))
    }

    pub fn rename_template_collection(
        &self,
        slug: &str,
        new_name: &str,
    ) -> Result<flow_storage::TemplateCollectionRecord, AppError> {
        templates::validate_slug(slug).map_err(|e| AppError::Template(e.to_string()))?;
        if new_name.trim().is_empty() {
            return Err(AppError::Template(
                "collection display name cannot be empty".into(),
            ));
        }
        self.store
            .rename_template_collection(slug, new_name)
            .map_err(|e| AppError::Template(e.to_string()))
    }

    pub fn delete_template_collection(&self, slug: &str) -> Result<(), AppError> {
        templates::validate_slug(slug).map_err(|e| AppError::Template(e.to_string()))?;
        self.store
            .delete_template_collection(slug)
            .map_err(|e| AppError::Template(e.to_string()))
    }

    /// Browse the Template Hub catalog (downloadable flow templates).
    pub fn list_template_hub(&self) -> Vec<template_hub::TemplateHubEntry> {
        template_hub::catalog()
    }

    /// Same as [`Self::list_template_hub`] but annotated with where each
    /// entry is installed locally + whether the catalog ships a newer
    /// version than what the user has on disk. Used by the Hub UI to render
    /// the "Installed in <collection>" indicator and the "Update available"
    /// pill.
    pub fn list_template_hub_with_status(
        &self,
    ) -> Result<Vec<template_hub::TemplateHubEntryWithStatus>, AppError> {
        template_hub::catalog_with_status(&self.store, &Self::templates_dir())
            .map_err(|e| AppError::Template(e.to_string()))
    }

    /// Install a Hub catalog entry into the user-chosen collection. Refuses
    /// if a template with the catalog slug already exists locally (re-add
    /// must go through [`Self::update_hub_template`]).
    pub fn add_hub_template(
        &self,
        hub_slug: &str,
        collection_slug: &str,
    ) -> Result<TemplateRecord, AppError> {
        template_hub::add_to_collection(
            &Self::templates_dir(),
            &self.store,
            hub_slug,
            collection_slug,
        )
        .map_err(|e| AppError::Template(e.to_string()))
    }

    /// Overwrite an installed Hub template in place with the catalog's
    /// current graph + version. When `force` is false and the user has
    /// edited the file since install, the call refuses so the frontend can
    /// confirm before clobbering local edits.
    pub fn update_hub_template(
        &self,
        hub_slug: &str,
        force: bool,
    ) -> Result<TemplateRecord, AppError> {
        template_hub::update_installed(&Self::templates_dir(), &self.store, hub_slug, force)
            .map_err(|e| AppError::Template(e.to_string()))
    }

    /// Browse the Node Hub catalog. Sorted by `sortKey` then slug.
    pub fn list_node_catalog(&self) -> Vec<nodes::NodeCatalogEntry> {
        node_hub::catalog()
    }

    /// Same as [`Self::list_node_catalog`] but annotated with install
    /// status (which entries are present in `node_library` and whether
    /// the catalog ships a newer version).
    pub fn list_node_catalog_with_status(
        &self,
    ) -> Result<Vec<node_hub::NodeCatalogEntryWithStatus>, AppError> {
        node_hub::catalog_with_status(&self.store)
            .map_err(|e| AppError::Template(e.to_string()))
    }

    /// Runtime view: every installed kind's full scheme, read from disk.
    /// This is what the canvas, palette, factory, inspector, and renderer
    /// consume - the embedded catalog only participates in installation.
    pub fn list_installed_nodes(&self) -> Result<Vec<nodes::NodeCatalogEntry>, AppError> {
        nodes::list_installed_schemes(&self.store, &Self::nodes_dir())
            .map_err(|e| AppError::Template(e.to_string()))
    }

    /// Install a catalog entry into the user library. Refuses on slug
    /// collision (use [`Self::update_installed_node`] to overwrite).
    pub fn add_node_to_library(
        &self,
        slug: &str,
    ) -> Result<flow_storage::NodeLibraryRow, AppError> {
        node_hub::add_to_library(&Self::nodes_dir(), &self.store, slug)
            .map_err(|e| AppError::Template(e.to_string()))
    }

    /// Overwrite an installed scheme with the catalog's current version
    /// and bump `node_library.version`.
    pub fn update_installed_node(
        &self,
        slug: &str,
        force: bool,
    ) -> Result<flow_storage::NodeLibraryRow, AppError> {
        node_hub::update_installed(&Self::nodes_dir(), &self.store, slug, force)
            .map_err(|e| AppError::Template(e.to_string()))
    }

    /// Remove an installed scheme from disk and the `node_library` row.
    pub fn uninstall_node(&self, slug: &str) -> Result<(), AppError> {
        node_hub::uninstall(&Self::nodes_dir(), &self.store, slug)
            .map_err(|e| AppError::Template(e.to_string()))
    }

    /// Resolve a catalog entry by slug. Prefers the on-disk installed
    /// scheme so user edits round-trip; falls back to the embedded
    /// catalog (the operator-supplied default seed data).
    pub fn get_node_catalog_entry(
        &self,
        slug: &str,
    ) -> Result<nodes::NodeCatalogEntry, AppError> {
        if let Ok(entry) = nodes::read_scheme(&Self::nodes_dir(), slug) {
            return Ok(entry);
        }
        nodes::entry_by_slug(slug)
            .ok_or_else(|| AppError::Template(format!("node `{slug}` not found")))
    }

    /// Parse a DSL document into a `FlowGraph`. Positions in the result are
    /// always `(0.0, 0.0)` - callers (the desktop UI) run an auto-layout pass
    /// before mounting on the canvas.
    pub fn import_dsl(&self, source: &str) -> Result<FlowGraph, AppError> {
        flow_dsl::parse(source).map_err(|e| AppError::Dsl {
            line: e.line,
            col: e.col,
            message: e.message,
        })
    }

    /// Serialize a `FlowGraph` to canonical DSL text.
    pub fn serialize_dsl(&self, graph: &FlowGraph) -> String {
        flow_dsl::serialize(graph)
    }

    fn settings_path() -> PathBuf {
        let mut path = Self::flow_dir();
        path.push("settings.json");
        path
    }

    pub fn settings(&self) -> Settings {
        self.settings.snapshot()
    }

    pub fn update_settings(&self, patch: SettingsPatch) -> Result<Settings, AppError> {
        self.settings
            .update(patch)
            .map_err(|e| AppError::Settings(e.to_string()))
    }

    pub fn list_cloud_providers(&self) -> Vec<flow_adapter_ai::registry::ProviderInfo> {
        let discovered_local = self.settings.local_ai_models();
        self.cloud_providers
            .iter()
            .map(|p| {
                // For the local provider, prefer the models discovered from
                // the running server (cached in settings) over the static
                // fallback, so the node inspector's dropdown reflects what's
                // actually loaded.
                let default_models = if p.category()
                    == flow_adapter_ai::ProviderCategory::Local
                    && !discovered_local.is_empty()
                {
                    discovered_local.clone()
                } else {
                    p.default_models().iter().map(|s| s.to_string()).collect()
                };
                flow_adapter_ai::registry::ProviderInfo {
                    name: p.name().to_string(),
                    env_var: p.env_var().to_string(),
                    default_models,
                    model_capabilities: p.model_capabilities(),
                    category: p.category(),
                }
            })
            .collect()
    }

    fn build_cloud_registry_for_run(&self) -> Arc<CloudAiRegistry> {
        let mut reg = CloudAiRegistry::new();
        for p in &self.cloud_providers {
            if self.settings.provider_enabled(p.name()) {
                reg.register(p.clone());
            }
        }
        Arc::new(reg)
    }

    /// Best-effort discovery of a `llama-server` (llama.cpp) executable so the
    /// user usually doesn't have to pick it manually. Checks `$PATH` then a few
    /// common install locations (Homebrew, /usr/local, /usr/bin).
    pub fn detect_llama_server() -> Option<String> {
        let names: &[&str] = if cfg!(windows) {
            &["llama-server.exe", "llama-cpp-server.exe"]
        } else {
            &["llama-server", "llama-cpp-server"]
        };
        let mut dirs: Vec<PathBuf> = Vec::new();
        if let Ok(path) = std::env::var("PATH") {
            dirs.extend(std::env::split_paths(&path));
        }
        for d in ["/opt/homebrew/bin", "/usr/local/bin", "/usr/bin"] {
            dirs.push(PathBuf::from(d));
        }
        for dir in dirs {
            for name in names {
                let candidate = dir.join(name);
                if candidate.is_file() {
                    return Some(candidate.to_string_lossy().to_string());
                }
            }
        }
        None
    }

    /// `<engines_dir()>/llama-server[.exe]` if a managed engine was already
    /// fetched, else `None`. Read-only - never triggers a fetch.
    fn managed_engine_path() -> Option<String> {
        let name = if cfg!(windows) {
            "llama-server.exe"
        } else {
            "llama-server"
        };
        let path = Self::engines_dir().join(name);
        path.is_file().then(|| path.to_string_lossy().to_string())
    }

    /// Resolve the `llama-server` binary the Hub would use: a previously-fetched
    /// managed engine, else the saved setting, else an auto-detected one.
    /// `None` means none is present yet (Load will fetch one).
    ///
    /// A saved setting that points at a now-missing file (e.g. an old build
    /// path that was cleaned away) is ignored so resolution falls through to a
    /// managed engine / `$PATH` instead of returning a dead path.
    pub fn llama_server_path(&self) -> Option<String> {
        Self::managed_engine_path()
            .or_else(|| Self::existing_binary(self.settings.llama_server_binary()))
            .or_else(Self::detect_llama_server)
    }

    /// Keep a binary path only if it points at an existing file; otherwise
    /// `None`. Used to drop stale saved/explicit `llama-server` paths so a
    /// missing file never short-circuits auto-resolution.
    fn existing_binary(path: Option<String>) -> Option<String> {
        path.filter(|p| !p.trim().is_empty())
            .filter(|p| std::path::Path::new(p).is_file())
    }

    /// Ensure a managed engine is present under `engines_dir()`, fetching it on
    /// first use. Fetch progress is mapped onto the shared `InstallProgress`
    /// pipeline (`stage: "engine"`) so callers surface it exactly like a hub
    /// model download.
    pub async fn ensure_llama_engine(
        &self,
        progress: Option<ProgressCallback>,
    ) -> Result<PathBuf, AppError> {
        let dir = Self::engines_dir();
        crate::llm_server::fetch::ensure_engine(&dir, move |p| {
            if let Some(cb) = &progress {
                cb(InstallProgress {
                    stage: "engine",
                    current: p.current,
                    total: p.total.unwrap_or(0),
                    message: p.message,
                });
            }
        })
        .await
        .map_err(AppError::AiInvocation)
    }

    /// Start the managed local LLM server (`llama-server`) for the given
    /// binary + model, falling back to the saved setting then auto-detection
    /// when omitted. On success: persist the paths, point `local_ai_base_url`
    /// at the managed endpoint, and enable local AI - so the existing `local`
    /// provider, flow-graph generator, and agent immediately use it.
    pub async fn start_local_llm(
        &self,
        binary: Option<String>,
        model: Option<String>,
        params: crate::llm_server::LlamaParams,
    ) -> Result<crate::llm_server::LlmServerStatus, AppError> {
        // Resolution order (managed engine wins, auto-fetch is the last resort):
        // explicit arg → fetched engine → saved setting → $PATH → fetch on demand.
        // Explicit/saved paths are dropped when the file is missing so a stale
        // path (e.g. a cleaned-up build dir) self-heals to a managed engine
        // instead of failing with "binary not found".
        let binary = match Self::existing_binary(binary)
            .or_else(Self::managed_engine_path)
            .or_else(|| Self::existing_binary(self.settings.llama_server_binary()))
            .or_else(Self::detect_llama_server)
        {
            Some(b) => b,
            None => self
                .ensure_llama_engine(None)
                .await?
                .to_string_lossy()
                .to_string(),
        };
        let model = model
            .filter(|s| !s.trim().is_empty())
            .or_else(|| self.settings.llama_server_model())
            .ok_or_else(|| AppError::AiInvocation("no LLM selected".into()))?;

        // Per-model load params are owned by the frontend (persisted in
        // `settings.llama_params` keyed by model id) and passed in here.
        let endpoint = self
            .llm_server
            .start(PathBuf::from(&binary), PathBuf::from(&model), params)
            .await
            .map_err(AppError::AiInvocation)?;

        // Discover the served model id(s) from `/v1/models` so the node
        // inspector + flow-graph-generator dropdowns reflect what's actually loaded
        // (this replaces the role the manual "Test connection" step played).
        // Best-effort: the server just answered readiness, so this normally
        // succeeds; fall back to an empty list on error.
        let discovered = flow_adapter_ai::LocalOpenAiProvider::new()
            .list_models(&endpoint)
            .await
            .unwrap_or_default();

        let _ = self.update_settings(SettingsPatch {
            llama_server_binary: Some(binary),
            llama_server_model: Some(model),
            local_ai_base_url: Some(endpoint),
            local_ai_models: Some(discovered),
            allow_local_ai: Some(true),
            ..Default::default()
        });
        Ok(self.llm_server.status())
    }

    /// Stop the managed local LLM server (idempotent). Clears the managed
    /// wiring (`local_ai_base_url` + discovered models) so the local provider,
    /// flow-graph generator, and agent don't keep calling a now-dead port.
    pub async fn stop_local_llm(&self) -> crate::llm_server::LlmServerStatus {
        self.llm_server.stop().await;
        let _ = self.update_settings(SettingsPatch {
            local_ai_base_url: Some(String::new()),
            local_ai_models: Some(Vec::new()),
            ..Default::default()
        });
        self.llm_server.status()
    }

    /// Status snapshot of the managed local LLM server.
    pub fn local_llm_status(&self) -> crate::llm_server::LlmServerStatus {
        self.llm_server.status()
    }

    pub fn store(&self) -> &Store {
        &self.store
    }

    pub fn list_executions(&self, limit: usize) -> Result<Vec<ExecutionRecord>, AppError> {
        self.store
            .list_executions(limit)
            .map_err(|e| AppError::Storage(e.to_string()))
    }

    pub fn get_execution(
        &self,
        execution_id: &str,
    ) -> Result<
        Option<(ExecutionRecord, Vec<ExecutionStepRecord>, Vec<InterceptionRecord>)>,
        AppError,
    > {
        self.store
            .get_execution(execution_id)
            .map_err(|e| AppError::Storage(e.to_string()))
    }

    /// Record a monitor interception (auto-fix) against a run, so the History
    /// view can show what was changed/added beyond the per-node steps.
    pub fn record_interception(
        &self,
        execution_id: String,
        failed_node: String,
        issue: Option<String>,
        summary: String,
        dsl_before: Option<String>,
        dsl_after: Option<String>,
    ) -> Result<(), AppError> {
        self.store
            .record_interception(&InterceptionRecord {
                execution_id,
                at: chrono::Utc::now(),
                failed_node,
                issue,
                summary,
                dsl_before,
                dsl_after,
            })
            .map_err(|e| AppError::Storage(e.to_string()))
    }

    pub async fn execute_graph(
        &self,
        graph: FlowGraph,
        events: Arc<dyn EventSink>,
    ) -> Result<ExecutionSummary, AppError> {
        self.execute_graph_with_id(Uuid::new_v4().to_string(), graph, events)
            .await
    }

    /// Run `graph` under a caller-supplied execution id. A fresh per-run
    /// [`flow_execution::RunControl`] is registered under that id for the
    /// lifetime of the run (so [`FlowApp::pause_run`] / [`resume_run`] /
    /// [`cancel_run`] can target this run while others keep running) and removed
    /// on completion. The host generates the id up front so it can map the run
    /// to the originating tab/flow before the `Started` event is observed.
    pub async fn execute_graph_with_id(
        &self,
        execution_id: String,
        graph: FlowGraph,
        events: Arc<dyn EventSink>,
    ) -> Result<ExecutionSummary, AppError> {
        self.run_graph_internal(execution_id, graph, events, None, None, None)
            .await
    }

    /// Shared run path behind [`execute_graph_with_id`] and [`run_agent_turn`].
    /// `confirm_destructive_override` forces the per-step destructive-action
    /// confirmation gate on/off (`None` follows the persisted setting).
    /// `edit_staging` is `None` for a normal run; an agent turn passes a buffer
    /// so fs mutations are staged for review instead of written to disk.
    /// `workspace_override` pins the execution workspace for this run (the
    /// autonomous loop passes the user's chosen folder); `None` resolves it
    /// from the stored per-flow path or the edition default.
    async fn run_graph_internal(
        &self,
        execution_id: String,
        graph: FlowGraph,
        events: Arc<dyn EventSink>,
        confirm_destructive_override: Option<bool>,
        edit_staging: Option<flow_execution::StagedEdits>,
        workspace_override: Option<PathBuf>,
    ) -> Result<ExecutionSummary, AppError> {
        let allow_cloud_ai = self.settings.allow_cloud_ai();
        let allow_local_ai = self.settings.allow_local_ai();
        let local_ai_base_url = self.settings.local_ai_base_url();
        let cloud_providers = self.build_cloud_registry_for_run();
        // The per-step confirmation gate follows the persisted setting, unless a
        // headless runner has forced it off (no way to confirm).
        let confirm_destructive = confirm_destructive_override.unwrap_or_else(|| {
            self.settings.confirm_destructive()
                && !self
                    .confirm_gate_disabled
                    .load(std::sync::atomic::Ordering::SeqCst)
        });
        // Each run gets its own control, registered under its id so the host can
        // steer this run independently of any other in-flight runs.
        let control: flow_execution::SharedRunControl = Default::default();
        if let Ok(mut runs) = self.runs.lock() {
            runs.insert(execution_id.clone(), control.clone());
        }
        // Trace the node lifecycle to the terminal / log file (under the
        // `flow_execution` target) by wrapping the host's sink, so a dev run
        // shows what the engine is doing - not just llama-server output.
        let events: Arc<dyn EventSink> = Arc::new(flow_execution::TracingSink::new(events));
        // The shared execution workspace for this flow; every shell/cli/fs node
        // defaults its cwd / workspaceRoot to it. Created up front so fs nodes
        // (which require an existing root) work against the scratch default.
        let workspace_root =
            workspace_override.unwrap_or_else(|| self.resolve_workspace(&graph.id));
        if let Err(e) = std::fs::create_dir_all(&workspace_root) {
            tracing::warn!(error = %e, path = %workspace_root.display(), "failed to create workspace dir");
        }
        let executor = Executor {
            adapters: self.adapters.clone(),
            sanitizer: self.sanitizer.clone(),
            events,
            cloud_providers,
            credentials: self.resolver.clone(),
            allow_cloud_ai,
            allow_local_ai,
            local_ai_base_url,
            stream_sink: self.stream_sink(),
            working_memory: self.working_memory.clone(),
            control,
            confirm_destructive,
            // The review gate needs a host UI to resolve it; headless runners
            // disable the confirmation gate, which marks exactly that absence.
            review_gate_available: !self
                .confirm_gate_disabled
                .load(std::sync::atomic::Ordering::SeqCst),
            edit_staging,
            workspace_root,
        };
        let result = executor.run_with_id(execution_id.clone(), &graph).await;
        if let Ok(mut runs) = self.runs.lock() {
            runs.remove(&execution_id);
        }
        // Persist working memory so set-variable values survive across sessions.
        self.persist_memory();
        result.map_err(|e| AppError::Execution(e.to_string()))
    }

    /// Run one coding-agent turn as a flow. Same executor path as
    /// [`execute_graph_with_id`] - events stream and the run is steerable
    /// (pause/resume/cancel) under `execution_id` - but filesystem mutations are
    /// *staged* (computed, not written). Returns the run summary plus the
    /// proposed edits for the IDE to review and apply.
    pub async fn run_agent_turn(
        &self,
        execution_id: String,
        graph: FlowGraph,
        events: Arc<dyn EventSink>,
    ) -> Result<AgentTurn, AppError> {
        let staging: flow_execution::StagedEdits = Default::default();
        let summary = self
            .run_graph_internal(execution_id, graph, events, None, Some(staging.clone()), None)
            .await?;
        let edits = staging.lock().map(|s| s.edits()).unwrap_or_default();
        Ok(AgentTurn { summary, edits })
    }

    /// Request a pause of the run with `execution_id`. Honored at the next node
    /// boundary; an in-flight node finishes first. No-op if no such run.
    pub fn pause_run(&self, execution_id: &str) {
        if let Ok(runs) = self.runs.lock() {
            if let Some(c) = runs.get(execution_id) {
                c.pause();
            }
        }
    }

    /// Release the paused run with `execution_id` so scheduling continues.
    pub fn resume_run(&self, execution_id: &str) {
        if let Ok(runs) = self.runs.lock() {
            if let Some(c) = runs.get(execution_id) {
                c.resume();
            }
        }
    }

    /// Resolve the AI review gate of the run with `execution_id`: record the
    /// human verdict and release the pause (RAO human authority). Approve
    /// passes the contract-bound output downstream; reject fails the node onto
    /// its fallback path. No-op if no such run.
    pub fn resolve_ai_review(&self, execution_id: &str, approved: bool) {
        if let Ok(runs) = self.runs.lock() {
            if let Some(c) = runs.get(execution_id) {
                c.resolve_review(approved);
            }
        }
    }

    /// Pause every active run. Convenience for headless runners (CLI / TUI)
    /// that only ever drive a single run and don't track its id.
    pub fn pause_all_runs(&self) {
        if let Ok(runs) = self.runs.lock() {
            for c in runs.values() {
                c.pause();
            }
        }
    }

    /// Resume every paused run. Companion to [`FlowApp::pause_all_runs`].
    pub fn resume_all_runs(&self) {
        if let Ok(runs) = self.runs.lock() {
            for c in runs.values() {
                c.resume();
            }
        }
    }

    /// Force the per-step destructive-action confirmation gate off for this
    /// instance. Headless runners (CLI / TUI) call this at startup: they have
    /// no confirmation UI, so a destructive node would otherwise block the run
    /// forever waiting on a confirm that can't arrive.
    pub fn disable_confirmation_gate(&self) {
        self.confirm_gate_disabled
            .store(true, std::sync::atomic::Ordering::SeqCst);
    }

    /// Cancel the run with `execution_id`. The current node finishes, remaining
    /// nodes are skipped, and the run ends with `cancelled` status. No-op if no
    /// such run.
    pub fn cancel_run(&self, execution_id: &str) {
        if let Ok(runs) = self.runs.lock() {
            if let Some(c) = runs.get(execution_id) {
                c.cancel();
            }
        }
    }

    /// Cancel every active run. Used by headless runners (CLI / TUI) which only
    /// ever have a single run and don't track its id, and as a stop-all signal.
    pub fn cancel_all_runs(&self) {
        if let Ok(runs) = self.runs.lock() {
            for c in runs.values() {
                c.cancel();
            }
        }
    }

    /// Clear session working memory. Call when starting a fresh agent session
    /// so `{{memory.<key>}}` doesn't leak values from a prior, unrelated run.
    pub fn clear_working_memory(&self) {
        if let Ok(mut mem) = self.working_memory.lock() {
            mem.clear();
        }
        let _ = self.store.clear_memory();
    }

    /// Persist the current working memory so `{{memory.<key>}}` survives across
    /// sessions (durable agent memory, roadmap E1). Best-effort.
    fn persist_memory(&self) {
        if let Ok(mem) = self.working_memory.lock() {
            let _ = self.store.save_memory(&mem);
        }
    }

    /// Static pre-apply verification of a proposed flow's DSL: returns advisories
    /// (destructive operations, possible sandbox escapes) for the review +
    /// interception modals. Best-effort - DSL that fails to parse yields no
    /// warnings (the caller surfaces the parse error separately).
    pub fn verify_flow(&self, dsl: &str) -> Vec<FlowWarning> {
        match flow_dsl::parse(dsl) {
            Ok(graph) => flow_execution::verify_graph(&graph),
            Err(_) => Vec::new(),
        }
    }

    /// Generate Flow DSL via a cloud AI provider (Claude / OpenAI / Gemini) or
    /// the local LLM server. Pick a model, hit Generate, and the frontend gets
    /// a parse-validated DSL string back regardless of source.
    ///
    /// The flow-graph-generator system prompt is shipped via each provider's
    /// native system channel (Anthropic top-level `system`, OpenAI
    /// system-role message, Gemini `system_instruction`) rather than
    /// concatenated into the user message. The PII sanitizer is
    /// intentionally bypassed: filenames and dataset names need to reach
    /// the model verbatim so the generated DSL refers to the right
    /// artifacts.
    ///
    /// **Parse validation with one retry.** The first response is parsed
    /// via `flow_dsl::parse`. On parse failure we re-issue the request
    /// with a corrective user message that quotes the bad DSL plus the
    /// parser's `line:col` and asks for a fix; the retry uses the same
    /// model, temperature, and max-tokens. Only parse-validated DSL is
    /// ever returned. If both attempts fail, the error message carries
    /// both `line:col` positions so the caller (and `generate_dsl_auto`)
    /// can surface them.
    pub async fn generate_dsl_via_cloud(
        &self,
        provider: &str,
        model: &str,
        prompt: &str,
        max_tokens: Option<u32>,
        temperature: Option<f32>,
    ) -> Result<String, AppError> {
        // The `local` provider is an on-device OpenAI-compatible server, not
        // network egress - it has its own gate (`allow_local_ai`), needs no
        // API key, and carries its endpoint via `base_url`.
        let is_local = provider == "local";
        if is_local {
            if !self.settings.allow_local_ai() {
                return Err(AppError::AiInvocation(
                    "local AI is disabled (Settings -> LLM -> Local -> enable)".into(),
                ));
            }
        } else if !self.settings.allow_cloud_ai() {
            return Err(AppError::AiInvocation(
                "cloud AI is disabled (Settings -> LLM -> Cloud -> enable to allow outbound calls)"
                    .into(),
            ));
        }
        let registry = self.build_cloud_registry_for_run();
        let p = registry.get(provider).ok_or_else(|| {
            AppError::AiInvocation(format!("provider '{provider}' is not enabled in Settings"))
        })?;
        let env_var = p.env_var();
        let api_key = match self.resolver.resolve_cloud_ai_key(provider, env_var) {
            Ok(k) => k,
            // Local servers usually need no auth; a missing key is fine.
            Err(_) if is_local => String::new(),
            Err(e) => {
                return Err(AppError::AiInvocation(format!(
                    "no API key for '{provider}': {e}. Set it in Settings or export {env_var}."
                )))
            }
        };
        let base_url = if is_local {
            self.settings.local_ai_base_url()
        } else {
            None
        };

        // Local "thinking" models (Qwen3, R1-distills) can spend thousands of
        // tokens in `reasoning_content` before emitting the answer; with a low
        // cap they get cut off mid-thought and return empty content. Local
        // inference is free, so floor the budget high.
        //
        // Cloud generation inlines file `content`, so a multi-file scaffold
        // easily exceeds a small cap and gets cut off mid-string - which
        // surfaces as a misleading `unterminated string`. A ceiling costs
        // nothing unless used (billing is per actual output token), so floor
        // cloud generation to the same headroom rather than trusting a small
        // caller default. Truncation past this floor is detected below.
        let effective_max_tokens = Some(max_tokens.unwrap_or(0).max(8192));
        let max_tokens = effective_max_tokens;

        let system_prompt = DSL_GENERATION_SYSTEM_PROMPT;

        tracing::info!(
            provider,
            model,
            is_local,
            prompt_chars = prompt.len(),
            "generating Flow DSL"
        );

        // Round 1: clean system + user split.
        let sink = self.stream_sink();
        let (raw1, finish1) = cloud_call_once(
            p.as_ref(),
            model,
            system_prompt,
            prompt,
            max_tokens,
            temperature,
            &api_key,
            base_url.clone(),
            sink.as_ref(),
        )
        .await?;
        let extracted1 = extract_dsl_text(&raw1);

        // Corrective retry loop. Each pass inspects the failure kind and sends
        // the matching corrective prompt: parse errors quote the parser's
        // `line:col`; semantic errors quote the specific contract violations.
        // Handling both kinds in one loop is what lets a parse fix that then
        // trips semantics (e.g. a `utility` node carrying `run-command`) still
        // get a semantic-correction pass instead of failing outright.
        const MAX_CORRECTIVE_RETRIES: usize = 3;
        let mut current = extracted1;
        let mut attempt = 0usize;
        let mut last_reason = match validate_generated_dsl(&current) {
            Ok(dsl) => {
                tracing::info!(provider, model, attempt, "Flow DSL generated on first try");
                return Ok(dsl);
            }
            Err(reason) => reason,
        };
        // A response cut off at the token limit yields an unterminated string no
        // amount of re-prompting can fix at the same budget, so short-circuit
        // with a clear, actionable error instead of burning corrective retries.
        if is_truncated(&finish1) {
            return Err(truncation_error(provider, model, max_tokens, &last_reason));
        }
        for _ in 0..MAX_CORRECTIVE_RETRIES {
            attempt += 1;
            let retry_user = if last_reason.starts_with("DSL parse error") {
                let parse_err = flow_dsl::parse(&current).expect_err("parse error expected");
                tracing::warn!(
                    provider,
                    attempt,
                    line = parse_err.line,
                    col = parse_err.col,
                    message = %parse_err.message,
                    "generated DSL failed to parse; retrying with corrective prompt"
                );
                build_retry_prompt(prompt, &current, &parse_err)
            } else {
                let graph = flow_dsl::parse(&current).expect("already parsed");
                let issues = dsl_semantics::validate_generated_graph(&graph);
                tracing::warn!(
                    provider,
                    attempt,
                    %last_reason,
                    "generated DSL failed semantic checks; retrying with corrective prompt"
                );
                build_semantic_retry_prompt(prompt, &current, &issues)
            };
            let (raw, finish) = cloud_call_once(
                p.as_ref(),
                model,
                system_prompt,
                &retry_user,
                max_tokens,
                temperature,
                &api_key,
                base_url.clone(),
                sink.as_ref(),
            )
            .await?;
            current = extract_dsl_text(&raw);
            match validate_generated_dsl(&current) {
                Ok(dsl) => {
                    tracing::info!(
                        provider,
                        model,
                        attempt,
                        "Flow DSL generated after corrective retry"
                    );
                    return Ok(dsl);
                }
                Err(reason) => last_reason = reason,
            }
            // Same token-limit short-circuit as round 1: don't keep retrying a
            // response the model could not finish within the budget.
            if is_truncated(&finish) {
                return Err(truncation_error(provider, model, max_tokens, &last_reason));
            }
        }

        // Retries exhausted. For a trailing parse error, append the targeted
        // hint (adapter-name-as-node-kind, malformed string, …) so the surfaced
        // error tells the user how to fix it, not just where it broke.
        let hint = last_reason
            .strip_prefix("DSL parse error at ")
            .and_then(|rest| rest.split_once(" - "))
            .and_then(|(_, msg)| parse_error_hint(msg))
            .map(|h| format!("\n\n{h}"))
            .unwrap_or_default();
        tracing::warn!(
            provider,
            model,
            reason = %last_reason,
            dsl = %truncate_for_log(&current),
            "Flow DSL generation failed after corrective retries"
        );
        Err(AppError::AiInvocation(format!(
            "{provider} model returned invalid DSL after {MAX_CORRECTIVE_RETRIES} corrective retries: {last_reason}{hint}"
        )))
    }

    /// Generate Flow DSL via a cloud/local-provider model and parse-validate
    /// it. `cloud` is `(provider, model_id)` - the loaded LLM is the `local`
    /// provider. Returns parse-valid DSL or an error.
    pub async fn generate_dsl_auto(
        &self,
        _local_model_id: Option<&str>,
        cloud: Option<(&str, &str)>,
        prompt: &str,
    ) -> Result<DslGenerationOutcome, AppError> {
        if let Some((provider, model)) = cloud {
            let dsl = self
                .generate_dsl_via_cloud(provider, model, prompt, Some(2048), None)
                .await
                .map_err(|e| AppError::AiInvocation(format!("cloud: {e}")))?;
            return Ok(DslGenerationOutcome {
                dsl,
                source: DslSource::Cloud {
                    provider: provider.to_string(),
                    model_id: model.to_string(),
                },
                fallback_reason: None,
                plan: None,
            });
        }
        Err(AppError::AiInvocation(
            "no flow-graph generator configured (select a model or provider)".into(),
        ))
    }

    /// Agentic generation: generate DSL from the prompt alone, then ask a
    /// chat-capable model for a one-paragraph plan describing the result.
    /// Generation never reads local/user data - the workspace is an execution
    /// concern resolved at run time, not a generation input (see ADR-0007).
    /// The plan call is best-effort: failures (no chat-capable provider
    /// reachable, network error, empty response) leave `plan = None` so
    /// the caller can fall back to a deterministic client-side synthesis.
    ///
    /// "Chat-capable provider" means anything we can call via
    /// `generate_dsl_via_cloud` with an arbitrary system prompt - the
    /// loaded LLM (when `cloud == Some(("local", _))`) or a real cloud
    /// provider. With neither available there is no chat path; the call
    /// returns `Ok(outcome)` with `plan = None`.
    pub async fn agentic_generate(
        &self,
        local_model_id: Option<&str>,
        cloud: Option<(&str, &str)>,
        prompt: &str,
    ) -> Result<DslGenerationOutcome, AppError> {
        let mut outcome = self.generate_dsl_auto(local_model_id, cloud, prompt).await?;
        outcome.plan = self.synthesize_plan(&outcome.dsl, cloud).await;
        Ok(outcome)
    }

    /// Observe-and-re-plan: regenerate the workflow given the prior DSL and a
    /// feedback string describing what went wrong when it ran (from
    /// [`build_loop_observations`]). The original task `prompt` is preserved so
    /// the model keeps the goal in view while fixing the failures. Goes through
    /// the same prompt-only agentic pipeline as a first generation, just with an
    /// augmented prompt.
    ///
    /// This is the single re-plan step. The frontend's review-then-run path
    /// calls it per user request; [`Self::agentic_run_loop`] calls it
    /// repeatedly for the autonomous path.
    pub async fn agentic_replan(
        &self,
        local_model_id: Option<&str>,
        cloud: Option<(&str, &str)>,
        prompt: &str,
        prior_dsl: &str,
        observations: &str,
    ) -> Result<DslGenerationOutcome, AppError> {
        let augmented = format!(
            "{prompt}\n\n\
             You previously generated this Flow DSL:\n\n{prior_dsl}\n\n\
             When it ran, these problems occurred:\n\n{observations}\n\n\
             Produce a corrected Flow DSL that fixes the failing steps. Keep the \
             parts that already succeeded. Emit DSL only."
        );
        self.agentic_generate(local_model_id, cloud, &augmented)
            .await
    }

    /// Autonomous agentic loop (the no-human-in-the-loop path): repeatedly
    /// generate → run → observe → re-plan until a run has zero failures or the
    /// [`MAX_AGENTIC_ITERATIONS`] safety cap is hit. Each iteration's run emits
    /// the normal `flow:execution` events through `events` (so the Log Panel
    /// streams live); the returned [`AgenticLoopSummary`] carries the final DSL
    /// for the caller to load onto the canvas plus a per-iteration record.
    ///
    /// Safety: bounded by the iteration cap, and every generated graph still
    /// runs through `execute_graph`, so the existing per-adapter sandbox and
    /// `allow_cloud_ai` / `allow_local_ai` gates apply unchanged - autonomy
    /// does not bypass them.
    pub async fn agentic_run_loop(
        &self,
        local_model_id: Option<&str>,
        cloud: Option<(&str, &str)>,
        prompt: &str,
        workspace: Option<&str>,
        events: Arc<dyn EventSink>,
    ) -> Result<AgenticLoopSummary, AppError> {
        // `workspace` pins where each iteration's run executes - it is never an
        // input to generation (ADR-0007).
        let ws_override = workspace
            .filter(|w| !w.trim().is_empty())
            .map(expand_home);
        let mut steps = Vec::new();
        let mut prev_dsl: Option<String> = None;
        let mut observations: Option<String> = None;
        let mut last_outcome: Option<DslGenerationOutcome> = None;
        let cap = self.settings.max_agentic_iterations();
        let budget_secs = self.settings.max_agentic_seconds();
        let token_budget = self.settings.max_agentic_tokens();
        let started = std::time::Instant::now();
        let mut tokens_used: u64 = 0;

        for iter in 0..cap {
            // Run budgets: stop starting new iterations once the wall-clock or
            // cumulative-token ceiling is hit (the first iteration always runs).
            // `0` = unlimited for either.
            let over_time =
                budget_secs > 0 && started.elapsed().as_secs() >= budget_secs as u64;
            let over_tokens = token_budget > 0 && tokens_used >= token_budget as u64;
            if iter > 0 && (over_time || over_tokens) {
                let last = last_outcome.expect("loop body runs at least once");
                return Ok(AgenticLoopSummary {
                    iterations: iter,
                    status: "budget_exhausted".into(),
                    final_dsl: last.dsl,
                    final_plan: last.plan,
                    steps,
                });
            }
            let outcome = match (&prev_dsl, &observations) {
                (Some(dsl), Some(obs)) => {
                    self.agentic_replan(local_model_id, cloud, prompt, dsl, obs)
                        .await?
                }
                _ => self.agentic_generate(local_model_id, cloud, prompt).await?,
            };

            let graph = flow_dsl::parse(&outcome.dsl).map_err(|e| {
                AppError::AiInvocation(format!(
                    "generated DSL failed to parse at {}:{} - {}",
                    e.line, e.col, e.message
                ))
            })?;

            // Run with a capturing sink alongside the caller's so we can read
            // per-node outcomes back for the next re-plan.
            let capture = CapturingSink::default();
            let sink: Arc<dyn EventSink> =
                Arc::new(MultiSink::new(vec![events.clone(), Arc::new(capture.clone())]));
            let summary = self
                .run_graph_internal(
                    Uuid::new_v4().to_string(),
                    graph,
                    sink,
                    None,
                    None,
                    ws_override.clone(),
                )
                .await?;
            let captured = capture.events.lock().map(|g| g.clone()).unwrap_or_default();
            tokens_used = tokens_used.saturating_add(sum_ai_tokens(&captured));

            let converged = summary.failed == 0;
            let obs = if converged {
                None
            } else {
                Some(build_loop_observations(&captured, &summary))
            };
            steps.push(AgenticLoopStep {
                iteration: iter,
                run_status: summary.status.clone(),
                succeeded: summary.succeeded,
                failed: summary.failed,
                skipped: summary.skipped,
                observations: obs.clone(),
            });

            if converged {
                return Ok(AgenticLoopSummary {
                    iterations: iter + 1,
                    status: "succeeded".into(),
                    final_dsl: outcome.dsl,
                    final_plan: outcome.plan,
                    steps,
                });
            }

            observations = obs;
            prev_dsl = Some(outcome.dsl.clone());
            last_outcome = Some(outcome);
        }

        // Cap hit without a clean run. Return the last attempt so the user can
        // inspect/repair it manually.
        let last = last_outcome.expect("loop body runs at least once");
        Ok(AgenticLoopSummary {
            iterations: cap,
            status: "exhausted".into(),
            final_dsl: last.dsl,
            final_plan: last.plan,
            steps,
        })
    }

    /// Best-effort second pass: ask a chat-capable provider to summarise
    /// the generated DSL. Returns `Some(text)` on
    /// success; `None` on any failure (gating off, no provider, parse
    /// error, empty body). Callers must not treat `None` as fatal.
    async fn synthesize_plan(
        &self,
        dsl: &str,
        cloud: Option<(&str, &str)>,
    ) -> Option<String> {
        // We can only synthesise a plan via a chat model - a cloud provider
        // or the loaded local LLM (the `local` provider).
        let (provider, model) = cloud?;
        let registry = self.build_cloud_registry_for_run();
        let p = registry.get(provider)?;
        let is_local = provider == "local";
        let allowed = if is_local {
            self.settings.allow_local_ai()
        } else {
            self.settings.allow_cloud_ai()
        };
        if !allowed {
            return None;
        }
        let env_var = p.env_var();
        let api_key = match self.resolver.resolve_cloud_ai_key(provider, env_var) {
            Ok(k) => k,
            Err(_) if is_local => String::new(),
            Err(_) => return None,
        };
        let base_url = if is_local {
            self.settings.local_ai_base_url()
        } else {
            None
        };
        let system = "You explain a Flow DSL workflow for a \
                      reviewer about to apply it. Write one short paragraph \
                      (under 60 words) summarising what the workflow does, \
                      then a numbered list with one line per step describing \
                      that step's purpose. No code fences, no DSL syntax.";
        let user = format!(
            "Flow DSL to summarise:\n\n{dsl}\n\nReturn the paragraph + numbered list only."
        );
        let req = flow_adapter_ai::CloudAiRequest {
            model: model.to_string(),
            prompt: user,
            // Plan calls don't need a giant budget; keep it tight so even
            // a tiny local model returns quickly.
            max_tokens: Some(if is_local { 1024 } else { 600 }),
            temperature: Some(0.2),
            api_key,
            system: Some(system.to_string()),
            base_url,
            stream: true,
            call_id: Some(format!("plan-{}", uuid::Uuid::new_v4())),
            ..Default::default()
        };
        let sink = self.stream_sink();
        match p.invoke_stream(&req, sink.as_ref()).await {
            Ok(resp) => {
                let trimmed = resp.text.trim().to_string();
                if trimmed.is_empty() {
                    None
                } else {
                    Some(trimmed)
                }
            }
            Err(e) => {
                tracing::debug!(provider, error = %e, "plan synthesis call failed; returning None");
                None
            }
        }
    }

    pub async fn load_graph(&self, graph_id: &str) -> Result<FlowGraph, AppError> {
        if graph_id.is_empty() {
            return Err(AppError::GraphNotFound("empty graph id".into()));
        }

        Ok(FlowGraph {
            subflows: Vec::new(),
            id: graph_id.to_string(),
            name: "JCL Pipeline".into(),
            version: "0.1.0".into(),
            description: None,
            nodes: vec![
                FlowNode {
                    id: "validate-jcl".into(),
                    node_type: "ai".into(),
                    position: Position { x: 80.0, y: 80.0 },
                    data: serde_json::json!({
                        "label": "Review JCL",
                        "modelId": "local-llm",
                        "input": "{{input}}"
                    }),
                },
                FlowNode {
                    id: "submit-jcl".into(),
                    node_type: "action".into(),
                    position: Position { x: 380.0, y: 80.0 },
                    data: serde_json::json!({
                        "label": "Submit JCL",
                        "actionId": "cli-tool",
                        "adapter": "cli",
                        "bin": "zowe",
                        "command": "jobs submit local-file"
                    }),
                },
            ],
            edges: vec![FlowEdge {
                id: "e1".into(),
                source: "validate-jcl".into(),
                target: "submit-jcl".into(),
                label: None,
                condition: None,
                outcome: flow_domain::graph::EdgeOutcome::Always,
            }],
        })
    }

    pub async fn validate_graph(&self, graph: FlowGraph) -> Result<(), AppError> {
        if graph.nodes.is_empty() {
            return Err(AppError::Validation(
                "graph must contain at least one node".into(),
            ));
        }
        Ok(())
    }

    pub async fn run_flow(&self, _req: ExecutionRequest) -> Result<ExecutionResult, AppError> {
        Ok(ExecutionResult {
            execution_id: Uuid::new_v4().to_string(),
            status: ExecutionStatus::Succeeded,
            started_at: Utc::now(),
            ended_at: Some(Utc::now()),
        })
    }
}