use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use std::process::Command;
use std::time::Duration;

use anyhow::{Context, Result, bail};
use beam_core::{FinalOutputKind, InitConfig};
use serde::Deserialize;
use serde::de::DeserializeOwned;
use serde_json::Value;

use crate::adapter::{OpenCodeState, PollResult, SpawnSpec, SubmitResult};
use crate::backend::SessionBackend;

const OPENCODE_CURSOR_LOOKBACK_MS: u64 = 5_000;

#[derive(Debug, Clone, Deserialize)]
pub struct OpenCodeTranscriptSource {
    pub db_path: PathBuf,
    pub session_id: String,
}

#[derive(Debug, Clone, Deserialize)]
struct OpenCodeSessionRow {
    id: String,
    time_updated: Option<u64>,
}

#[derive(Debug, Clone, Deserialize)]
struct OpenCodeMessageRow {
    message_id: String,
    session_id: String,
    message_time_created: Option<u64>,
    message_time_updated: Option<u64>,
    message_data: String,
    part_id: Option<String>,
    part_time_updated: Option<u64>,
    part_data: Option<String>,
}

#[derive(Debug, Clone)]
struct GroupedMessage {
    id: String,
    session_id: String,
    time_created: u64,
    time_updated: u64,
    data: Value,
    parts: Vec<GroupedPart>,
}

#[derive(Debug, Clone)]
struct GroupedPart {
    time_updated: u64,
    data: Value,
}

pub fn create_state(init: &InitConfig) -> OpenCodeState {
    let home = std::env::var("HOME").unwrap_or_default();
    let data_dir = PathBuf::from(format!("{}/.local/share/opencode", home));
    let expected_session_id = init.cli_session_id.clone();
    OpenCodeState {
        data_dir,
        expected_session_id,
        working_dir: init.working_dir.clone(),
        cli_session_id: init.cli_session_id.clone(),
        transcript_offset: 0,
        emitted_final_text: None,
    }
}

pub fn build_spawn_spec(_state: &OpenCodeState, init: &InitConfig) -> SpawnSpec {
    let mut args = Vec::new();
    if let Some(model) = &init.model {
        if !model.is_empty() {
            args.push("--model".to_string());
            args.push(model.clone());
        }
    }
    if let Some(prompt) = &init.initial_prompt {
        args.push("--prompt".to_string());
        args.push(prompt.clone());
    }
    args.extend(init.cli_args.clone());
    SpawnSpec {
        bin: init.cli_bin.clone(),
        args,
    }
}

pub async fn write_input(
    state: &mut OpenCodeState,
    backend: &dyn SessionBackend,
    content: &str,
) -> Result<SubmitResult> {
    let Some(source) = wait_for_source(state).await else {
        return Ok(SubmitResult {
            submitted: false,
            cli_session_id: state.cli_session_id.clone(),
            failure_reason: Some("OpenCode transcript source not found".to_string()),
        });
    };
    let base_offset = current_opencode_session_offset(&source)?;
    state
        .cli_session_id
        .get_or_insert_with(|| source.session_id.clone());

    backend.send_text(content).await?;
    tokio::time::sleep(Duration::from_millis(200)).await;
    backend.send_enter().await?;
    for attempt in 0..4 {
        tokio::time::sleep(Duration::from_millis(800)).await;
        if opencode_submit_confirmed(&source, base_offset, content)? {
            return Ok(SubmitResult {
                submitted: true,
                cli_session_id: state.cli_session_id.clone(),
                ..Default::default()
            });
        }
        if attempt < 3 {
            backend.send_enter().await?;
        }
    }
    Ok(SubmitResult {
        submitted: false,
        cli_session_id: state.cli_session_id.clone(),
        failure_reason: Some("OpenCode transcript did not confirm submit".to_string()),
    })
}

pub fn poll(state: &mut OpenCodeState) -> Result<PollResult> {
    let source = current_source(state);
    let Some(source) = source else {
        return Ok(PollResult {
            cli_session_id: state.cli_session_id.clone(),
            ..Default::default()
        });
    };

    let drain = drain_opencode_session(&source, state.transcript_offset)?;
    state.transcript_offset = drain.new_offset;
    if state.cli_session_id.is_none() {
        state.cli_session_id = Some(source.session_id.clone());
    }

    let mut result = PollResult {
        cli_session_id: state.cli_session_id.clone(),
        ..Default::default()
    };

    for event in drain.events {
        if event.kind != "assistant_final" {
            continue;
        }
        if !event.text.is_empty() && state.emitted_final_text.as_deref() != Some(&event.text) {
            state.emitted_final_text = Some(event.text.clone());
            result.final_output = Some(event.text);
            result.final_output_kind = Some(FinalOutputKind::Bridge);
            result.prompt_ready = true;
        }
    }

    Ok(result)
}

pub fn opencode_db_candidates(data_dir: &Path) -> Vec<PathBuf> {
    let Ok(entries) = std::fs::read_dir(data_dir) else {
        return Vec::new();
    };
    entries
        .filter_map(|entry| entry.ok())
        .filter_map(|entry| {
            let path = entry.path();
            let name = path.file_name()?.to_str()?;
            if !name.starts_with("opencode") || !name.ends_with(".db") {
                return None;
            }
            match entry.file_type() {
                Ok(ft) if ft.is_file() => Some(path),
                _ => None,
            }
        })
        .collect()
}

pub fn find_opencode_session_by_id(
    session_id: Option<&str>,
    db_paths: &[PathBuf],
) -> Option<OpenCodeTranscriptSource> {
    let session_id = session_id?;
    for db_path in db_paths {
        if !db_path.exists() {
            continue;
        }
        if let Ok(Some(row)) = query_session_by_id(db_path, session_id) {
            return Some(OpenCodeTranscriptSource {
                db_path: db_path.clone(),
                session_id: row.id,
            });
        }
    }
    None
}

pub fn find_latest_opencode_session_by_directory(
    directory: Option<&str>,
    db_paths: &[PathBuf],
) -> Option<OpenCodeTranscriptSource> {
    let directory = directory?;
    let mut best: Option<(PathBuf, OpenCodeSessionRow)> = None;
    for db_path in db_paths {
        if !db_path.exists() {
            continue;
        }
        if let Ok(Some(row)) = query_latest_session_by_directory(db_path, directory) {
            let replace = match &best {
                None => true,
                Some((_, current)) => {
                    row.time_updated.unwrap_or(0) > current.time_updated.unwrap_or(0)
                }
            };
            if replace {
                best = Some((db_path.clone(), row));
            }
        }
    }
    best.map(|(db_path, row)| OpenCodeTranscriptSource {
        db_path,
        session_id: row.id,
    })
}

pub fn drain_opencode_session(
    source: &OpenCodeTranscriptSource,
    from_offset: u64,
) -> Result<OpenCodeDrainResult> {
    if !source.db_path.exists() {
        return Ok(OpenCodeDrainResult {
            events: Vec::new(),
            new_offset: from_offset,
        });
    }
    let rows = query_changed_rows(source, from_offset)?;
    let mut grouped: BTreeMap<String, GroupedMessage> = BTreeMap::new();
    let mut new_offset = from_offset;
    for row in rows {
        let entry = grouped
            .entry(row.message_id.clone())
            .or_insert_with(|| GroupedMessage {
                id: row.message_id.clone(),
                session_id: row.session_id.clone(),
                time_created: row.message_time_created.unwrap_or(0),
                time_updated: row.message_time_updated.unwrap_or(0),
                data: parse_object(&row.message_data),
                parts: Vec::new(),
            });
        if let Some(time_updated) = row.message_time_updated {
            entry.time_updated = entry.time_updated.max(time_updated);
        }
        if let (Some(_part_id), Some(part_data)) = (row.part_id, row.part_data) {
            entry.parts.push(GroupedPart {
                time_updated: row.part_time_updated.unwrap_or(0),
                data: parse_object(&part_data),
            });
        }
    }

    let mut messages = grouped.into_values().collect::<Vec<_>>();
    messages.sort_by(|a, b| {
        a.time_created
            .cmp(&b.time_created)
            .then_with(|| a.id.cmp(&b.id))
    });

    let mut events = Vec::new();
    for message in messages {
        new_offset = new_offset.max(message.time_updated).max(
            message
                .parts
                .iter()
                .map(|part| part.time_updated)
                .max()
                .unwrap_or(0),
        );
        let role = message.data.get("role").and_then(Value::as_str);
        match role {
            Some("user") => {
                let text = text_from_parts(&message.parts);
                if text.is_empty() {
                    continue;
                }
                events.push(OpenCodeBridgeEvent {
                    uuid: format!("opencode:{}:{}", source.db_path.display(), message.id),
                    timestamp_ms: message_timestamp_ms(&message, false),
                    kind: "user".to_string(),
                    text,
                    source_session_id: Some(message.session_id.clone()),
                });
            }
            Some("assistant") => {
                if should_skip_assistant(&message.data) {
                    continue;
                }
                let text = text_from_parts(&message.parts);
                if text.is_empty() {
                    continue;
                }
                events.push(OpenCodeBridgeEvent {
                    uuid: format!("opencode:{}:{}", source.db_path.display(), message.id),
                    timestamp_ms: message_timestamp_ms(&message, true),
                    kind: "assistant_final".to_string(),
                    text,
                    source_session_id: Some(message.session_id.clone()),
                });
            }
            _ => {}
        }
    }

    Ok(OpenCodeDrainResult { events, new_offset })
}

fn current_source(state: &OpenCodeState) -> Option<OpenCodeTranscriptSource> {
    let db_paths = opencode_db_candidates(&state.data_dir);
    if let Some(source) =
        find_opencode_session_by_id(state.expected_session_id.as_deref(), &db_paths)
    {
        return Some(source);
    }
    find_latest_opencode_session_by_directory(Some(&state.working_dir), &db_paths)
}

async fn wait_for_source(state: &OpenCodeState) -> Option<OpenCodeTranscriptSource> {
    for _ in 0..12 {
        if let Some(source) = current_source(state) {
            return Some(source);
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
    }
    current_source(state)
}

pub fn current_opencode_session_offset(source: &OpenCodeTranscriptSource) -> Result<u64> {
    if !source.db_path.exists() {
        return Ok(0);
    }
    let mut script = String::from(
        r#"
import sqlite3
conn = sqlite3.connect(__DB_PATH__)
row = conn.execute(
    """
    SELECT COALESCE(MAX(value), 0) FROM (
      SELECT time_updated AS value FROM message WHERE session_id = ?
      UNION ALL
      SELECT time_updated AS value FROM part WHERE session_id = ?
    )
    """,
    (__SESSION_ID__, __SESSION_ID__),
).fetchone()
print(row[0] or 0)
"#,
    );
    script = script.replace(
        "__DB_PATH__",
        &json_string(&source.db_path.display().to_string()),
    );
    script = script.replace("__SESSION_ID__", &json_string(&source.session_id));
    let proc = Command::new("python3")
        .args(["-c", &script])
        .output()
        .context("failed to query opencode session offset")?;
    if !proc.status.success() {
        bail!(
            "{}",
            String::from_utf8_lossy(&proc.stderr).trim().to_string()
        );
    }
    Ok(String::from_utf8_lossy(&proc.stdout)
        .trim()
        .parse::<u64>()
        .unwrap_or(0))
}

fn opencode_submit_confirmed(
    source: &OpenCodeTranscriptSource,
    from_offset: u64,
    expected_text: &str,
) -> Result<bool> {
    if !source.db_path.exists() {
        return Ok(false);
    }
    let drain = drain_opencode_session(source, from_offset)?;
    let prefix = expected_text.chars().take(40).collect::<String>();
    Ok(drain
        .events
        .iter()
        .any(|event| event.kind == "user" && event.text.starts_with(&prefix)))
}

fn should_skip_assistant(data: &Value) -> bool {
    data.get("error").is_some() && data.get("summary").is_none()
}

fn query_changed_rows(
    source: &OpenCodeTranscriptSource,
    offset: u64,
) -> Result<Vec<OpenCodeMessageRow>> {
    let lower_bound = offset.saturating_sub(OPENCODE_CURSOR_LOOKBACK_MS);
    let mut script = String::from(
        r#"
import json
import sqlite3
conn = sqlite3.connect(__DB_PATH__)
conn.row_factory = sqlite3.Row
rows = conn.execute(
    """
    WITH changed AS (
      SELECT m.id
      FROM message m
      LEFT JOIN part p ON p.message_id = m.id
      WHERE m.session_id = ?
        AND (m.time_updated > ? OR COALESCE(p.time_updated, 0) > ?)
      GROUP BY m.id
    )
    SELECT
      m.id AS message_id,
      m.session_id AS session_id,
      m.time_created AS message_time_created,
      m.time_updated AS message_time_updated,
      m.data AS message_data,
      p.id AS part_id,
      p.time_updated AS part_time_updated,
      p.data AS part_data
    FROM message m
    LEFT JOIN part p ON p.message_id = m.id
    WHERE m.id IN (SELECT id FROM changed)
    ORDER BY m.time_created, m.id, p.time_created, p.id
    """,
    (__SESSION_ID__, __LOWER_BOUND__, __LOWER_BOUND__),
).fetchall()
print(json.dumps([dict(r) for r in rows], ensure_ascii=False))
"#,
    );
    script = script.replace(
        "__DB_PATH__",
        &json_string(&source.db_path.display().to_string()),
    );
    script = script.replace("__SESSION_ID__", &json_string(&source.session_id));
    script = script.replace("__LOWER_BOUND__", &lower_bound.to_string());
    run_python_json(&script)
}

fn query_session_by_id(db_path: &Path, session_id: &str) -> Result<Option<OpenCodeSessionRow>> {
    let mut script = String::from(
        r#"
import json
import sqlite3
conn = sqlite3.connect(__DB_PATH__)
conn.row_factory = sqlite3.Row
row = conn.execute(
    "SELECT id, time_updated FROM session WHERE id = ? LIMIT 1",
    (__SESSION_ID__,),
).fetchone()
print(json.dumps(dict(row), ensure_ascii=False) if row else "null")
"#,
    );
    script = script.replace("__DB_PATH__", &json_string(&db_path.display().to_string()));
    script = script.replace("__SESSION_ID__", &json_string(session_id));
    run_python_json(&script)
}

fn query_latest_session_by_directory(
    db_path: &Path,
    directory: &str,
) -> Result<Option<OpenCodeSessionRow>> {
    let mut script = String::from(
        r#"
import json
import sqlite3
conn = sqlite3.connect(__DB_PATH__)
conn.row_factory = sqlite3.Row
row = conn.execute(
    """
    SELECT id, directory, time_updated
    FROM session
    WHERE directory = ?
    ORDER BY time_updated DESC
    LIMIT 1
    """,
    (__DIRECTORY__,),
).fetchone()
print(json.dumps(dict(row), ensure_ascii=False) if row else "null")
"#,
    );
    script = script.replace("__DB_PATH__", &json_string(&db_path.display().to_string()));
    script = script.replace("__DIRECTORY__", &json_string(directory));
    run_python_json(&script)
}

fn parse_object(raw: &str) -> Value {
    serde_json::from_str(raw).unwrap_or(Value::Null)
}

fn text_from_parts(parts: &[GroupedPart]) -> String {
    parts
        .iter()
        .filter_map(|part| {
            if part.data.get("type").and_then(Value::as_str) != Some("text") {
                return None;
            }
            if part.data.get("ignored").and_then(Value::as_bool) == Some(true) {
                return None;
            }
            part.data
                .get("text")
                .and_then(Value::as_str)
                .and_then(|text| {
                    if text.trim().is_empty() {
                        None
                    } else {
                        Some(text.to_string())
                    }
                })
        })
        .collect::<Vec<_>>()
        .join("")
}

fn number_value(value: &Value) -> Option<u64> {
    value
        .as_u64()
        .or_else(|| value.as_i64().and_then(|v| u64::try_from(v).ok()))
}

fn message_timestamp_ms(message: &GroupedMessage, assistant_final: bool) -> u64 {
    if let Some(time) = message.data.get("time").and_then(Value::as_object) {
        if assistant_final {
            if let Some(completed) = time.get("completed").and_then(number_value) {
                return completed;
            }
        }
        if let Some(created) = time.get("created").and_then(number_value) {
            return created;
        }
    }
    message.time_updated.max(message.time_created)
}

fn json_string(value: &str) -> String {
    serde_json::to_string(value).expect("json string")
}

fn run_python_json<T: DeserializeOwned>(script: &str) -> Result<T> {
    let proc = Command::new("python3")
        .args(["-c", script])
        .output()
        .context("failed to run python3")?;
    if !proc.status.success() {
        bail!(
            "{}",
            String::from_utf8_lossy(&proc.stderr).trim().to_string()
        );
    }
    let stdout = String::from_utf8_lossy(&proc.stdout).trim().to_string();
    if stdout.is_empty() {
        bail!("python3 returned empty output");
    }
    Ok(serde_json::from_str(&stdout)?)
}

#[derive(Debug, Clone)]
pub struct OpenCodeBridgeEvent {
    #[allow(dead_code)]
    pub uuid: String,
    #[allow(dead_code)]
    pub timestamp_ms: u64,
    pub kind: String,
    pub text: String,
    #[allow(dead_code)]
    pub source_session_id: Option<String>,
}

#[derive(Debug, Clone)]
pub struct OpenCodeDrainResult {
    pub events: Vec<OpenCodeBridgeEvent>,
    pub new_offset: u64,
}

#[cfg(test)]
mod tests {
    use super::*;
    use async_trait::async_trait;
    use std::fs;
    use std::process::Command;
    use std::sync::{Arc, Mutex};
    use uuid::Uuid;

    fn temp_dir(name: &str) -> PathBuf {
        std::env::temp_dir().join(format!("beam-opencode-{}-{}", name, Uuid::new_v4()))
    }

    fn create_test_db(db_path: &Path) {
        let mut script = String::from(
            r#"
import sqlite3
conn = sqlite3.connect(__DB_PATH__)
conn.executescript("""
CREATE TABLE session (
  id TEXT PRIMARY KEY,
  directory TEXT,
  time_updated INTEGER
);
CREATE TABLE message (
  id TEXT PRIMARY KEY,
  session_id TEXT,
  time_created INTEGER,
  time_updated INTEGER,
  data TEXT
);
CREATE TABLE part (
  id TEXT PRIMARY KEY,
  message_id TEXT,
  session_id TEXT,
  time_created INTEGER,
  time_updated INTEGER,
  data TEXT
);
""")
conn.execute(
    "INSERT INTO session (id, directory, time_updated) VALUES (?, ?, ?)",
    ("sess-1", "/repo/opencode", 1500),
)
conn.execute(
    "INSERT INTO message (id, session_id, time_created, time_updated, data) VALUES (?, ?, ?, ?, ?)",
    ("msg-user", "sess-1", 1000, 1001, '{"role":"user","id":"msg-user"}'),
)
conn.execute(
    "INSERT INTO part (id, message_id, session_id, time_created, time_updated, data) VALUES (?, ?, ?, ?, ?, ?)",
    ("part-user", "msg-user", "sess-1", 1002, 1002, '{"type":"text","text":"hello"}'),
)
conn.execute(
    "INSERT INTO message (id, session_id, time_created, time_updated, data) VALUES (?, ?, ?, ?, ?)",
    ("msg-asst", "sess-1", 1300, 1500, '{"role":"assistant","id":"msg-asst"}'),
)
conn.execute(
    "INSERT INTO part (id, message_id, session_id, time_created, time_updated, data) VALUES (?, ?, ?, ?, ?, ?)",
    ("part-step", "msg-asst", "sess-1", 1400, 1400, '{"type":"step-start"}'),
)
conn.execute(
    "INSERT INTO part (id, message_id, session_id, time_created, time_updated, data) VALUES (?, ?, ?, ?, ?, ?)",
    ("part-text", "msg-asst", "sess-1", 1490, 1490, '{"type":"text","text":"hi there"}'),
)
conn.commit()
"#,
        );
        script = script.replace("__DB_PATH__", &json_string(&db_path.display().to_string()));
        let status = Command::new("python3")
            .args(["-c", &script])
            .status()
            .expect("python3 available");
        assert!(status.success(), "failed to create sqlite db");
    }

    fn append_user_submit(db_path: &Path, text: &str, time_created: u64, time_updated: u64) {
        let message_id = format!("msg-{}", Uuid::new_v4());
        let part_id = format!("part-{}", Uuid::new_v4());
        let mut script = String::from(
            r#"
import sqlite3
conn = sqlite3.connect(__DB_PATH__)
conn.execute(
    "INSERT INTO message (id, session_id, time_created, time_updated, data) VALUES (?, ?, ?, ?, ?)",
    (__MESSAGE_ID__, __SESSION_ID__, __TIME_CREATED__, __TIME_UPDATED__, __MESSAGE_DATA__),
)
conn.execute(
    "INSERT INTO part (id, message_id, session_id, time_created, time_updated, data) VALUES (?, ?, ?, ?, ?, ?)",
    (__PART_ID__, __MESSAGE_ID__, __SESSION_ID__, __PART_CREATED__, __PART_UPDATED__, __PART_DATA__),
)
conn.commit()
"#,
        );
        script = script.replace("__DB_PATH__", &json_string(&db_path.display().to_string()));
        script = script.replace("__MESSAGE_ID__", &json_string(&message_id));
        script = script.replace("__PART_ID__", &json_string(&part_id));
        script = script.replace("__SESSION_ID__", &json_string("sess-1"));
        script = script.replace("__TIME_CREATED__", &time_created.to_string());
        script = script.replace("__TIME_UPDATED__", &time_updated.to_string());
        script = script.replace(
            "__MESSAGE_DATA__",
            &json_string(r#"{"role":"user","id":"submit"}"#),
        );
        script = script.replace("__PART_CREATED__", &(time_created + 1).to_string());
        script = script.replace("__PART_UPDATED__", &time_updated.to_string());
        script = script.replace(
            "__PART_DATA__",
            &json_string(&format!(
                r#"{{"type":"text","text":{}}}"#,
                json_string(text)
            )),
        );
        let status = Command::new("python3")
            .args(["-c", &script])
            .status()
            .expect("python3 available");
        assert!(status.success(), "failed to append submit row");
    }

    #[derive(Clone, Default)]
    struct RecordingBackend {
        db_path: PathBuf,
        buffer: Arc<Mutex<String>>,
        append_on_enter: bool,
        next_time: Arc<Mutex<u64>>,
        calls: Arc<Mutex<Vec<String>>>,
    }

    impl RecordingBackend {
        fn new(db_path: PathBuf, append_on_enter: bool, start_time: u64) -> Self {
            Self {
                db_path,
                buffer: Arc::new(Mutex::new(String::new())),
                append_on_enter,
                next_time: Arc::new(Mutex::new(start_time)),
                calls: Arc::new(Mutex::new(Vec::new())),
            }
        }

        fn calls(&self) -> Vec<String> {
            self.calls.lock().unwrap().clone()
        }
    }

    #[async_trait]
    impl SessionBackend for RecordingBackend {
        async fn spawn(
            &mut self,
            _bin: &str,
            _args: &[String],
            _opts: crate::backend::SpawnOpts,
        ) -> Result<()> {
            Ok(())
        }

        async fn send_text(&self, text: &str) -> Result<()> {
            self.calls.lock().unwrap().push(format!("text:{text}"));
            self.buffer.lock().unwrap().push_str(text);
            Ok(())
        }

        async fn send_enter(&self) -> Result<()> {
            self.calls.lock().unwrap().push("enter".to_string());
            if self.append_on_enter {
                let content = {
                    let mut buffer = self.buffer.lock().unwrap();
                    let content = buffer.clone();
                    buffer.clear();
                    content
                };
                if !content.is_empty() {
                    let mut next_time = self.next_time.lock().unwrap();
                    let created = *next_time + 1;
                    let updated = created + 1;
                    *next_time = updated;
                    append_user_submit(&self.db_path, &content, created, updated);
                }
            }
            Ok(())
        }

        async fn send_special_keys(&self, _keys: &[String]) -> Result<()> {
            Ok(())
        }

        async fn paste_text(&self, text: &str) -> Result<()> {
            self.send_text(text).await
        }

        async fn write_raw(&self, _text: &str) -> Result<()> {
            Ok(())
        }

        async fn raw_input(&self, _text: &str) -> Result<()> {
            Ok(())
        }

        async fn capture_viewport(&self) -> Result<String> {
            Ok(String::new())
        }

        async fn capture_current_screen(&self) -> Result<String> {
            Ok(String::new())
        }

        async fn is_alive(&self) -> Result<bool> {
            Ok(true)
        }

        async fn child_pid(&self) -> Result<Option<u32>> {
            Ok(None)
        }

        async fn kill(&mut self) -> Result<()> {
            Ok(())
        }

        async fn destroy_session(&mut self) -> Result<()> {
            Ok(())
        }

        async fn cursor_position(&self) -> Result<Option<(u16, u16)>> {
            Ok(None)
        }

        fn subscribe(&self) -> tokio::sync::broadcast::Receiver<String> {
            let (_tx, rx) = tokio::sync::broadcast::channel(1);
            rx
        }
    }

    #[test]
    fn opencode_reader_finds_sessions_and_final_output() {
        let root = temp_dir("poll");
        let data_dir = root.join("share").join("opencode");
        fs::create_dir_all(&data_dir).unwrap();
        let db_path = data_dir.join("opencode.db");
        create_test_db(&db_path);

        let candidates = opencode_db_candidates(&data_dir);
        assert_eq!(candidates, vec![db_path.clone()]);

        let source =
            find_opencode_session_by_id(Some("sess-1"), &candidates).expect("session lookup");
        assert_eq!(source.db_path, db_path);
        assert_eq!(source.session_id, "sess-1");

        let latest = find_latest_opencode_session_by_directory(Some("/repo/opencode"), &candidates)
            .expect("directory lookup");
        assert_eq!(latest.session_id, "sess-1");

        let drain = drain_opencode_session(&source, 0).expect("drain");
        assert_eq!(drain.events.len(), 2);
        assert_eq!(drain.events[0].kind, "user");
        assert_eq!(drain.events[0].text, "hello");
        assert_eq!(drain.events[1].kind, "assistant_final");
        assert_eq!(drain.events[1].text, "hi there");

        let mut state = OpenCodeState {
            data_dir: data_dir.clone(),
            expected_session_id: Some("sess-1".to_string()),
            working_dir: "/repo/opencode".to_string(),
            cli_session_id: None,
            transcript_offset: 0,
            emitted_final_text: None,
        };
        let first = poll(&mut state).expect("first poll");
        assert_eq!(first.final_output.as_deref(), Some("hi there"));
        assert_eq!(first.final_output_kind, Some(FinalOutputKind::Bridge));
        assert!(first.prompt_ready);
        assert_eq!(state.transcript_offset, 1500);
        let second = poll(&mut state).expect("second poll");
        assert!(second.final_output.is_none());
        assert!(second.prompt_ready == false);
        let _ = fs::remove_dir_all(root);
    }

    #[test]
    fn opencode_reader_dedupes_repeat_final_output_and_recovers_offset() {
        let root = temp_dir("dedupe");
        let data_dir = root.join("share").join("opencode");
        fs::create_dir_all(&data_dir).unwrap();
        let db_path = data_dir.join("opencode.db");
        create_test_db(&db_path);

        let mut state = OpenCodeState {
            data_dir: data_dir.clone(),
            expected_session_id: Some("sess-1".to_string()),
            working_dir: "/repo/opencode".to_string(),
            cli_session_id: None,
            transcript_offset: 0,
            emitted_final_text: None,
        };
        let first = poll(&mut state).expect("first poll");
        assert_eq!(first.final_output.as_deref(), Some("hi there"));
        assert_eq!(state.transcript_offset, 1500);

        append_user_submit(&db_path, "hello opencode", 1600, 1601);
        let second = poll(&mut state).expect("second poll");
        assert!(second.final_output.is_none());

        let mut script = String::from(
            r#"
import sqlite3
conn = sqlite3.connect(__DB_PATH__)
conn.executescript("""
DELETE FROM part;
DELETE FROM message;
""")
conn.execute(
    "INSERT INTO message (id, session_id, time_created, time_updated, data) VALUES (?, ?, ?, ?, ?)",
    ("msg-user-2", "sess-1", 2000, 2001, '{"role":"user","id":"msg-user-2"}'),
)
conn.execute(
    "INSERT INTO part (id, message_id, session_id, time_created, time_updated, data) VALUES (?, ?, ?, ?, ?, ?)",
    ("part-user-2", "msg-user-2", "sess-1", 2002, 2002, '{"type":"text","text":"fresh"}'),
)
conn.execute(
    "INSERT INTO message (id, session_id, time_created, time_updated, data) VALUES (?, ?, ?, ?, ?)",
    ("msg-asst-2", "sess-1", 2100, 2200, '{"role":"assistant","id":"msg-asst-2"}'),
)
conn.execute(
    "INSERT INTO part (id, message_id, session_id, time_created, time_updated, data) VALUES (?, ?, ?, ?, ?, ?)",
    ("part-asst-2", "msg-asst-2", "sess-1", 2190, 2190, '{"type":"text","text":"after truncate"}'),
)
conn.commit()
"#,
        );
        script = script.replace("__DB_PATH__", &json_string(&db_path.display().to_string()));
        let status = Command::new("python3")
            .args(["-c", &script])
            .status()
            .expect("python3 available");
        assert!(status.success(), "failed to rewrite sqlite db");

        let third = poll(&mut state).expect("third poll");
        assert_eq!(third.final_output.as_deref(), Some("after truncate"));
        assert_eq!(third.final_output_kind, Some(FinalOutputKind::Bridge));
        assert_eq!(state.transcript_offset, 2200);
        let _ = fs::remove_dir_all(root);
    }

    #[tokio::test]
    async fn opencode_write_input_verifies_transcript_before_reporting_success() {
        let root = temp_dir("submit");
        let data_dir = root.join("share").join("opencode");
        fs::create_dir_all(&data_dir).unwrap();
        let db_path = data_dir.join("opencode.db");
        let mut script = String::from(
            r#"
import sqlite3
conn = sqlite3.connect(__DB_PATH__)
conn.executescript("""
CREATE TABLE session (
  id TEXT PRIMARY KEY,
  directory TEXT,
  time_updated INTEGER
);
CREATE TABLE message (
  id TEXT PRIMARY KEY,
  session_id TEXT,
  time_created INTEGER,
  time_updated INTEGER,
  data TEXT
);
CREATE TABLE part (
  id TEXT PRIMARY KEY,
  message_id TEXT,
  session_id TEXT,
  time_created INTEGER,
  time_updated INTEGER,
  data TEXT
);
""")
conn.execute(
    "INSERT INTO session (id, directory, time_updated) VALUES (?, ?, ?)",
    ("sess-1", "/repo/opencode", 1000),
)
conn.commit()
"#,
        );
        script = script.replace("__DB_PATH__", &json_string(&db_path.display().to_string()));
        let status = Command::new("python3")
            .args(["-c", &script])
            .status()
            .expect("python3 available");
        assert!(status.success(), "failed to create sqlite db");

        let mut state = OpenCodeState {
            data_dir: data_dir.clone(),
            expected_session_id: Some("sess-1".to_string()),
            working_dir: "/repo/opencode".to_string(),
            cli_session_id: None,
            transcript_offset: 0,
            emitted_final_text: None,
        };
        let backend = RecordingBackend::new(db_path.clone(), true, 1000);
        let result = write_input(&mut state, &backend, "hello opencode")
            .await
            .expect("write input");
        assert!(result.submitted);
        assert_eq!(result.cli_session_id.as_deref(), Some("sess-1"));
        assert!(backend.calls().iter().any(|call| call == "enter"));
        let _ = fs::remove_dir_all(root);
    }

    #[tokio::test]
    async fn opencode_write_input_reports_failure_when_transcript_does_not_confirm() {
        let root = temp_dir("submit-fail");
        let data_dir = root.join("share").join("opencode");
        fs::create_dir_all(&data_dir).unwrap();
        let db_path = data_dir.join("opencode.db");
        let mut script = String::from(
            r#"
import sqlite3
conn = sqlite3.connect(__DB_PATH__)
conn.executescript("""
CREATE TABLE session (
  id TEXT PRIMARY KEY,
  directory TEXT,
  time_updated INTEGER
);
CREATE TABLE message (
  id TEXT PRIMARY KEY,
  session_id TEXT,
  time_created INTEGER,
  time_updated INTEGER,
  data TEXT
);
CREATE TABLE part (
  id TEXT PRIMARY KEY,
  message_id TEXT,
  session_id TEXT,
  time_created INTEGER,
  time_updated INTEGER,
  data TEXT
);
""")
conn.execute(
    "INSERT INTO session (id, directory, time_updated) VALUES (?, ?, ?)",
    ("sess-1", "/repo/opencode", 1000),
)
conn.commit()
"#,
        );
        script = script.replace("__DB_PATH__", &json_string(&db_path.display().to_string()));
        let status = Command::new("python3")
            .args(["-c", &script])
            .status()
            .expect("python3 available");
        assert!(status.success(), "failed to create sqlite db");

        let mut state = OpenCodeState {
            data_dir: data_dir.clone(),
            expected_session_id: Some("sess-1".to_string()),
            working_dir: "/repo/opencode".to_string(),
            cli_session_id: None,
            transcript_offset: 0,
            emitted_final_text: None,
        };
        let backend = RecordingBackend::new(db_path.clone(), false, 1000);
        let result = write_input(&mut state, &backend, "hello opencode")
            .await
            .expect("write input");
        assert!(!result.submitted);
        assert!(
            result
                .failure_reason
                .as_deref()
                .unwrap_or("")
                .contains("did not confirm")
        );
        let _ = fs::remove_dir_all(root);
    }
}
