pond-db 0.16.0

//! oh-my-pi adapter (github.com/can1357/oh-my-pi, the `omp` binary).
//!
//! A pi fork that kept pi's v3 session record model, so this adapter is a thin
//! shell over the shared v3 codec in [`super::pi_coding_agent`]: the header's
//! `id` / `timestamp` / `cwd`, `message` rows carrying an `AgentMessage` under
//! `.message`, and the `parentId` -> `id` entry chain are all pi v3 verbatim.
//! What differs is the container, and only in two ways:
//!
//! - **The physical title slot.** Current omp files begin with a fixed-width
//!   256-byte `{"type":"title","v":1,...}` line and put the session header on
//!   the SECOND physical line (`session-title-slot.ts`); legacy files start at
//!   the header. omp's own loader strips the slot and folds it into the logical
//!   header, and so does this adapter - the slot's title / source / updatedAt
//!   land in `options.source.title_slot`, never dropped
//!   (spec.md#adapter-integrity-no-silent-drops) and never a message.
//! - **The sessions directory.** `~/.omp/agent/sessions/<bucket>/`, where the
//!   bucket is scope-encoded from the cwd (`session-paths.ts`): `-<home-relative>`
//!   under `$HOME`, `-tmp-<rel>` under the temp root, else pi's
//!   `--<encoded-absolute>--`. A hashed `<scope>-<basename>-<sha256>` form also
//!   exists in the wild - omp 17.2.5-17.2.8 wrote it before reverting - and omp
//!   migrates those dirs back into the encoded name on access. Inert either way:
//!   the project comes from the header's `cwd`, and the directory name is kept
//!   only as the placement hint every JSONL adapter records. Verified against
//!   omp 17.2.15 by reading what its own `SessionManager` writes.
//!
//! omp's entry taxonomy is much wider than pi's (`ttsr_injection`,
//! `credential_pin`, `session_init`, `mode_change`, `label`, `title_change`,
//! `reset_boundary`, `custom`, `custom_message`, ...) and needs no per-type work:
//! the shared v3 mapper turns any unrecognized record into a System carrier with
//! its raw record preserved.
//!
//! Two fidelity notes, both deliberate:
//!
//! - omp externalizes images to `~/.omp/agent/blobs/<sha256>` and stores
//!   `blob:sha256:<hash>` refs inline. pond ingests the ref, not the bytes - the
//!   transcript stays faithful to the file, and search never wanted the pixels.
//! - The header's `parentSession` FIELD is carried verbatim in `options.source`
//!   but NOT mapped to `parent_session_id`. omp writes either a session id or a
//!   session path there depending on the flow and its own docs call it metadata,
//!   not a typed key, so resolving it is a lineage feature for pi and omp
//!   together (`docs/plans/2608-06-pi-pond-plugin-and-v4-adapters-implementation-plan.md`
//!   section 9 item 2), not a guess this adapter makes per row
//!   (spec.md#model-no-synthesis). Sessions in an artifacts directory are the
//!   separate case - see [`link_subagent`].
//!
//! Ingest-only by design: see [`OhMyPiFactory::serialize`].
//!
//! Format watch: fixtures are generated by driving omp's OWN session code -
//! `tests/fixtures/adapter/oh-my-pi/generate-fixtures.mjs`, last run against
//! `@oh-my-pi/pi-coding-agent@17.3.4`. A diff against the committed fixtures on
//! a later omp is the signal that this adapter needs the same move. Regenerating
//! 17.2.15 -> 17.3.4 left all three original files byte-identical: the record
//! format and the slot serializer did not move, only the container did, which is
//! the artifacts directory [`link_subagent`] reads.

use std::path::{Path, PathBuf};

use serde_json::{Value, json};

use super::{
    Adapter, AdapterError, AdapterFactory, AdapterYieldStream, DiscoverFuture, Env,
    RestoreFidelity, RestoredFile, SkipOracle, SourceWatermark, config_path,
    jsonl::{
        BoundedRow, JsonlTree, jsonl_tree_discover, jsonl_tree_events, jsonl_tree_plan,
        peek_last_line, peek_nth_line,
    },
    pi_coding_agent::{SourcePlacement, row_timestamp, v3_events_from_row, v3_session_from_row},
};
use crate::{sessions::IngestEvent, wire::Session};

const NAME: &str = "oh-my-pi";

/// The only session version this adapter decodes. omp is at 3 and migrates
/// older files up on load, so a header naming anything higher is a format move
/// this build predates - a visible, counted skip rather than a half ingest.
///
/// Numerically equal to pi's `V3_FORMAT`, which is the tag the shared mapper
/// STAMPS, while this is the highest version this reader ACCEPTS. Same value
/// today, different jobs: if omp's record model ever moves off pi's, this one
/// rises and the stamped tag does not.
const SUPPORTED_SESSION_VERSION: i64 = 3;

/// The physical line the title slot occupies. omp writes it at byte 0 and
/// rewrites it in place, so framing is a position, not a shape that could
/// appear anywhere.
const FRAMING_LINE: usize = 1;

/// omp's title-slot record type (`session-title-slot.ts`).
const TITLE_SLOT_TYPE: &str = "title";

/// omp's fixed-width title slot shape marker (`session-title-slot.ts` writes
/// `v: 1` and its own parser rejects anything else). A different value is a
/// container move this build has not seen, not a slot to fold.
const SUPPORTED_TITLE_SLOT_VERSION: i64 = 1;

/// Why restore is refused, shared by the capability query and `serialize` so the
/// two surfaces can never drift.
const RESTORE_UNSUPPORTED: &str = "oh-my-pi captures sessions but cannot restore them yet; \
     `pond resume <id> --to pi-coding-agent --out-dir ~/.omp/agent` writes a v3 transcript \
     omp reads as a legacy session file - plus one per subagent, into the parent's \
     artifacts directory, since resume restores a whole lineage";

/// Stateless factory: opens [`OhMyPiAdapter`] instances and probes for omp's
/// canonical install location under `~/.omp/agent/sessions`.
pub struct OhMyPiFactory;

impl AdapterFactory for OhMyPiFactory {
    fn name(&self) -> &'static str {
        NAME
    }

    fn open(&self, config: Value) -> Result<Box<dyn Adapter>, AdapterError> {
        Ok(Box::new(OhMyPiAdapter::new(config_path(NAME, config)?)))
    }

    /// Home-relative on every platform, and field-verified: PR #147 synced 331
    /// omp sessions end-to-end on Windows 11 through this probe.
    fn probe_default(&self, env: &Env) -> Option<Value> {
        let path = env.home.join(".omp").join("agent").join("sessions");
        path.exists().then(|| json!({ "path": path }))
    }

    /// Ingest-only: omp restore is not implemented, and a restore pond cannot
    /// verify against a real omp is worse than no restore
    /// (spec.md#adapter-native-restore-lossless - fidelity is reported, never
    /// assumed). The reason names the working alternative rather than the
    /// symptom: omp loads a slot-less v3 file as a legacy session, which is
    /// exactly what the pi serializer emits.
    fn restore_unsupported(&self) -> Option<&'static str> {
        Some(RESTORE_UNSUPPORTED)
    }

    /// Unreachable through `pond resume`, which asks [`Self::restore_unsupported`]
    /// first. Kept honest rather than `unreachable!` so a future caller that
    /// skips the capability query gets an error, not a panic.
    fn serialize(
        &self,
        _session: &crate::sessions::SessionWithMessages,
        _fidelity: RestoreFidelity,
    ) -> Result<Vec<RestoredFile>, AdapterError> {
        Err(AdapterError::schema(NAME, NAME, RESTORE_UNSUPPORTED))
    }
}

/// Configured omp reader: a tree of `*.jsonl` session files under one sessions
/// root. omp's SQL / Redis `SessionStorage` backends publish through the same
/// local index, so there is no second source to open here.
#[derive(Debug, Clone)]
pub struct OhMyPiAdapter {
    root: PathBuf,
}

impl OhMyPiAdapter {
    pub fn new(root: impl Into<PathBuf>) -> Self {
        Self { root: root.into() }
    }
}

impl Adapter for OhMyPiAdapter {
    fn discover(&self) -> DiscoverFuture<'_> {
        jsonl_tree_discover(self)
    }

    fn events_with<'a>(&'a self, oracle: &'a dyn SkipOracle) -> AdapterYieldStream<'a> {
        jsonl_tree_events(self, oracle)
    }

    fn plan<'a>(&'a self, oracle: &'a dyn SkipOracle) -> super::PlanFuture<'a> {
        jsonl_tree_plan(self, oracle)
    }
}

impl JsonlTree for OhMyPiAdapter {
    type State = ();

    fn name(&self) -> &'static str {
        NAME
    }

    fn root(&self) -> &Path {
        &self.root
    }

    fn peek_session_id(&self, path: &Path, first_line: &str) -> Option<String> {
        let first: Value = serde_json::from_str(first_line).ok()?;
        session_id_from_head(path, first)
    }

    fn peek_watermark(&self, path: &Path) -> SourceWatermark {
        // Append-ordered, so the last line is the latest event. A file holding
        // only its slot and header has no ingestible content yet and reads as
        // `Opaque` (re-read) rather than fresh.
        //
        // The header needs an explicit guard because it carries a `timestamp`
        // that would read as an event time. The slot needs none: it records
        // `updatedAt`, and `row_timestamp` keys on `timestamp` alone, so a
        // slot-only file falls out as `Opaque` on its own.
        let last = || -> Option<i64> {
            let row: Value = serde_json::from_str(&peek_last_line(path)?).ok()?;
            if is_session_head(&row) {
                return None;
            }
            row_timestamp(&row).map(|ts| ts.timestamp_micros())
        };
        match last() {
            Some(ts) => SourceWatermark::At(ts),
            None => SourceWatermark::Opaque,
        }
    }

    fn unsupported_reason(&self, path: &Path, rows: &[BoundedRow]) -> Option<String> {
        // Three format moves, one visible outcome. Without this a moved
        // container falls through `session()` into `SkipReason::Empty`, which
        // renders as `empty` with NO reason and shrinks the progress
        // denominator - so an entire unreadable corpus would report as a clean
        // sync with nothing to do (spec.md#adapter-integrity-no-silent-drops).
        let unsupported = |what: String| {
            Some(format!(
                "{}: {what} is newer than this pond build understands; upgrade pond",
                path.display(),
            ))
        };

        let first = rows.first().map(|row| &row.value)?;
        if first.get("type").and_then(Value::as_str) == Some(TITLE_SLOT_TYPE)
            && first.get("v").and_then(Value::as_i64) != Some(SUPPORTED_TITLE_SLOT_VERSION)
        {
            let version = first.get("v").cloned().unwrap_or(Value::Null);
            return unsupported(format!(
                "oh-my-pi title-slot version {version} (supported: \
                 {SUPPORTED_TITLE_SLOT_VERSION})"
            ));
        }

        let head = &logical_head(rows)?.value;
        // A `kind: "header"` container is the harness-v2 move pi already made;
        // omp has not, and this reader decodes only the v3 record model.
        if head.get("kind").and_then(Value::as_str) == Some("header") {
            let version = head.get("version").cloned().unwrap_or(Value::Null);
            return unsupported(format!(
                "the oh-my-pi session container (`kind: \"header\"`, version {version})"
            ));
        }

        let version = head.get("version").and_then(Value::as_i64)?;
        (is_session_head(head) && version > SUPPORTED_SESSION_VERSION).then(|| {
            format!(
                "{}: oh-my-pi session version {version} is newer than this pond build \
                 understands (supported: {SUPPORTED_SESSION_VERSION}); upgrade pond",
                path.display(),
            )
        })
    }

    fn session(&self, path: &Path, rows: &[BoundedRow]) -> Result<Session, AdapterError> {
        let path_display = path.display().to_string();
        let head = logical_head(rows).ok_or_else(|| {
            // A slot with no header behind it is a session omp has not
            // materialized yet, not a decodable transcript.
            AdapterError::schema(
                NAME,
                path_display.clone(),
                "session file has a title slot but no session header",
            )
        })?;
        let at_head = format!("{path_display}:{}", head.line);
        if !is_session_head(&head.value) {
            return Err(AdapterError::schema(
                NAME,
                at_head,
                "first record after the optional title slot must be a `session` header",
            ));
        }

        let placement = SourcePlacement::from_path(path);
        let mut session = v3_session_from_row(NAME, &head.value, &at_head, &placement)?;
        fold_title_slot(&mut session, rows.first(), &at_head)?;
        link_subagent(&mut session, path, &self.root);
        Ok(session)
    }

    fn events_from_row(
        &self,
        session: &Session,
        row: &BoundedRow,
        _state: &mut Self::State,
    ) -> Result<Vec<IngestEvent>, String> {
        // The slot is container framing folded into the session, not an entry -
        // but only on the physical first line, which is the sole place omp
        // writes it. A `title`-shaped row anywhere else is not framing, so it
        // rides the carrier path and stays visible
        // (spec.md#adapter-integrity-no-silent-drops).
        if row.line == FRAMING_LINE && is_title_slot(&row.value) {
            return Ok(Vec::new());
        }
        v3_events_from_row(&session.id, row.line, &row.value, session.created_at)
    }
}

/// The header id of a session file, read through the optional title slot (which
/// holds no id, so the real header is one line further in current omp files).
fn session_id_from_head(path: &Path, first: Value) -> Option<String> {
    let head = if is_title_slot(&first) {
        peek_nth_line(path, 2).and_then(|line| serde_json::from_str::<Value>(&line).ok())?
    } else {
        first
    };
    if !is_session_head(&head) {
        return None;
    }
    head.get("id")
        .and_then(Value::as_str)
        .map(ToOwned::to_owned)
}

/// omp keeps a session's spawned children in an artifacts directory named after
/// its own file: `<dir>/<name>.jsonl` beside `<dir>/<name>/<AgentId>.jsonl`,
/// nesting the same way at any depth. A child header carries no parent field,
/// so the path is the only link (spec.md#model-no-synthesis).
///
/// The parent id is READ from that neighbouring header, never parsed out of the
/// directory name: agent ids keep `_` (omp sanitizes to `[A-Za-z0-9_-]`), so
/// splitting on it invents a parent for nested children and drops the link for
/// the rest. `source_agent` is the fixed `oh-my-pi/subagent` because
/// `generateTaskName()` would make the taxonomy unbounded - which also labels a
/// `/tan` fork a subagent, as omp's own exporter does.
fn link_subagent(session: &mut Session, path: &Path, root: &Path) {
    // A transcript directly at the configured root has no artifacts directory
    // above it, and looking for one would stat outside the tree pond was given.
    let Some(dir) = path.parent().filter(|dir| *dir != root) else {
        return;
    };
    let Some(dir_name) = dir.file_name().and_then(|name| name.to_str()) else {
        return;
    };
    let Some(parent_file) = dir.parent().map(|up| up.join(format!("{dir_name}.jsonl"))) else {
        return;
    };
    let Some(first) =
        peek_nth_line(&parent_file, 1).and_then(|line| serde_json::from_str::<Value>(&line).ok())
    else {
        return;
    };
    let Some(parent_id) = session_id_from_head(&parent_file, first) else {
        return;
    };
    session.parent_session_id = Some(parent_id);
    session.source_agent = format!("{NAME}/subagent");
}

/// omp's fixed-width first-line title slot (`session-title-slot.ts`): `type`
/// plus the shape marker, which is what separates it from a real entry.
fn is_title_slot(row: &Value) -> bool {
    row.get("type").and_then(Value::as_str) == Some(TITLE_SLOT_TYPE)
        && row.get("v").and_then(Value::as_i64) == Some(SUPPORTED_TITLE_SLOT_VERSION)
}

fn is_session_head(row: &Value) -> bool {
    row.get("type").and_then(Value::as_str) == Some("session")
}

/// The first row that is not the physical title slot - the logical head omp's
/// own loader exposes. One predicate, so the version gate and the session
/// mapper can never disagree about which row is the header.
fn logical_head(rows: &[BoundedRow]) -> Option<&BoundedRow> {
    rows.iter().find(|row| !is_title_slot(&row.value))
}

/// Carry the stripped slot into the session's source options so folding it is
/// lossless. Absent for a legacy slot-less file, which is why a missing slot is
/// not an error.
///
/// Keyed on the PHYSICAL first line, the same key
/// [`JsonlTree::events_from_row`] skips on, so the two can never disagree about
/// which row is framing: a slot anywhere else is left to the carrier path
/// instead of being both folded and carried.
fn fold_title_slot(
    session: &mut Session,
    first: Option<&BoundedRow>,
    at_head: &str,
) -> Result<(), AdapterError> {
    let Some(slot) = first
        .filter(|row| row.line == FRAMING_LINE && is_title_slot(&row.value))
        .map(|row| &row.value)
    else {
        return Ok(());
    };
    // `v3_session_from_row` always inserts `source` as an object, so this is an
    // invariant, not a fallible lookup - surfaced rather than skipped, because
    // silently returning here would drop the slot the caller just stripped.
    let source = session
        .options
        .get_mut("source")
        .and_then(Value::as_object_mut)
        .ok_or_else(|| {
            AdapterError::schema(
                NAME,
                at_head.to_owned(),
                "session options carry no `source` object to fold the title slot into",
            )
        })?;
    source.insert(
        "title_slot".to_owned(),
        json!({
            "title": slot.get("title"),
            "source": slot.get("source"),
            "updated_at": slot.get("updatedAt"),
        }),
    );
    Ok(())
}

#[cfg(test)]
mod tests {
    //! End-to-end tests for the oh-my-pi adapter: ingest the committed fixture
    //! corpus (generated by driving omp's own title-slot serializer, see
    //! `tests/fixtures/adapter/oh-my-pi/generate-fixtures.mjs`) and assert
    //! pond's canonical shape comes out the other side.
    #![allow(clippy::expect_used, clippy::unwrap_used)]

    use super::*;
    use crate::{
        adapter::jsonl::peek_first_line, handlers::ingest_adapter, sessions::Store, wire::Message,
    };
    use tempfile::TempDir;

    // Manifest-dir anchored: unit tests must not depend on the process cwd.
    const FIXTURES: &str = concat!(
        env!("CARGO_MANIFEST_DIR"),
        "/tests/fixtures/adapter/oh-my-pi/sessions"
    );
    const BUCKET: &str = "-Projects-omp-demo";
    const MAIN: &str = "0a1b2c3d4e5f6071";
    const FORK: &str = "1b2c3d4e5f607182";
    const LEGACY: &str = "legacy00000001";
    const SUBAGENT: &str = "2c3d4e5f60718293";
    const NESTED: &str = "3d4e5f60718293a4";
    const PROJECT: &str = "/Users/user/Projects/omp-demo";

    fn a_session() -> Session {
        Session {
            id: "sess1".to_owned(),
            parent_session_id: None,
            parent_message_id: None,
            source_agent: NAME.to_owned(),
            created_at: chrono::Utc::now(),
            project: crate::adapter::extract::Extracted::from_test_value(PROJECT.to_owned()),
            options: crate::wire::ProviderOptions::new(),
        }
    }

    fn main_file() -> PathBuf {
        Path::new(FIXTURES)
            .join(BUCKET)
            .join(format!("2026-08-12T10-00-00-000Z_{MAIN}.jsonl"))
    }

    /// The committed corpus carries the artifacts layout too, so the fixtures
    /// prove the linkage rather than only the hand-built shape above.
    #[tokio::test(flavor = "multi_thread")]
    async fn the_fixture_corpus_links_its_subagents() -> anyhow::Result<()> {
        let temp = TempDir::new()?;
        let store = ingest_fixtures(&temp).await?;

        let subagent = store
            .get_session(SUBAGENT)
            .await?
            .expect("subagent ingests");
        assert_eq!(subagent.session.parent_session_id.as_deref(), Some(MAIN));
        assert_eq!(subagent.session.source_agent, format!("{NAME}/subagent"));

        let nested = store.get_session(NESTED).await?.expect("nested ingests");
        assert_eq!(nested.session.parent_session_id.as_deref(), Some(SUBAGENT));

        let main = store.get_session(MAIN).await?.expect("main ingests");
        assert_eq!(main.session.parent_session_id, None);
        assert_eq!(main.session.source_agent, NAME);
        Ok(())
    }

    /// omp 17.3.4's `task` layout, captured on Windows 11: the subagent sits in
    /// a directory named after the parent's own session file and carries no
    /// parent field. Both halves matter - the child links back and drops out of
    /// default search, while the parent beside it stays top-level.
    #[tokio::test(flavor = "multi_thread")]
    async fn task_subagent_links_to_the_parent_named_by_its_directory() -> anyhow::Result<()> {
        let corpus = TempDir::new()?;
        // The bucket keeps an underscore: omp maps only `/`, `\` and `:`, so a
        // project named `omp_test` carries one straight into the directory name.
        let bucket = corpus.path().join("--C--dev-omp_test--");
        let parent_id = "01a000f8-ab2d-7000-903b-514cf631bc40";
        let child_id = "01a000f8-c2be-7000-85a6-eae2c18027a0";
        let grandchild_id = "01a000f8-d3cf-7000-91b7-6f0e2d5a4c81";
        let stem = format!("2026-08-14T15-51-31-885Z_{parent_id}");
        std::fs::create_dir_all(bucket.join(&stem))?;

        let transcript = |id: &str, ts: &str, text: &str| {
            format!(
                concat!(
                    r#"{{"type":"title","v":1,"title":"","updatedAt":"{ts}"}}"#,
                    "\n",
                    r#"{{"type":"session","version":3,"id":"{id}","timestamp":"{ts}","cwd":"C:\\dev\\omp-test"}}"#,
                    "\n",
                    r#"{{"type":"message","id":"m-{id}","timestamp":"{ts}","message":{{"role":"user","content":[{{"type":"text","text":"{text}"}}]}}}}"#,
                    "\n",
                ),
                id = id,
                ts = ts,
                text = text,
            )
        };
        std::fs::write(
            bucket.join(format!("{stem}.jsonl")),
            transcript(parent_id, "2026-08-14T15:51:31.885Z", "parent turn"),
        )?;
        // `code_review` is a legal agent id and splitting it on `_` would invent
        // the parent `review`; the grandchild below is where that used to bite.
        std::fs::create_dir_all(bucket.join(&stem).join("code_review"))?;
        std::fs::write(
            bucket.join(&stem).join("code_review.jsonl"),
            transcript(child_id, "2026-08-14T15:51:37.918Z", "subagent turn"),
        )?;
        std::fs::write(
            bucket.join(&stem).join("code_review").join("Bob.jsonl"),
            transcript(grandchild_id, "2026-08-14T15:51:44.010Z", "nested turn"),
        )?;

        let temp = TempDir::new()?;
        let store = Store::open_local(temp.path()).await?;
        ingest_adapter(
            &store,
            &OhMyPiAdapter::new(corpus.path()),
            &crate::adapter::NoopOracle,
            |_| {},
        )
        .await?;

        let child = store
            .get_session(child_id)
            .await?
            .expect("the subagent transcript ingests");
        assert_eq!(
            child.session.parent_session_id.as_deref(),
            Some(parent_id),
            "the parent id comes from the neighbouring file's header",
        );
        assert_eq!(child.session.source_agent, format!("{NAME}/subagent"));

        // Nesting is unconditional in omp - every subagent gets its own
        // artifacts directory on spawn - so the grandchild must link to the
        // subagent above it, not to an id parsed out of `code_review`.
        let grandchild = store
            .get_session(grandchild_id)
            .await?
            .expect("the nested transcript ingests");
        assert_eq!(
            grandchild.session.parent_session_id.as_deref(),
            Some(child_id),
        );
        assert_eq!(grandchild.session.source_agent, format!("{NAME}/subagent"));

        let parent = store
            .get_session(parent_id)
            .await?
            .expect("the parent transcript ingests");
        assert_eq!(parent.session.parent_session_id, None);
        assert_eq!(parent.session.source_agent, NAME);
        Ok(())
    }

    async fn ingest_fixtures(temp: &TempDir) -> anyhow::Result<Store> {
        let store = Store::open_local(temp.path().join("store")).await?;
        let summary = ingest_adapter(
            &store,
            &OhMyPiAdapter::new(FIXTURES),
            &crate::adapter::NoopOracle,
            |_| {},
        )
        .await?;
        anyhow::ensure!(summary.dropped_events == 0, "no event may be dropped");
        anyhow::ensure!(summary.dropped_sessions == 0, "no session may be dropped");
        Ok(store)
    }

    #[test]
    fn probe_default_finds_omp_sessions_under_home() -> anyhow::Result<()> {
        crate::adapter::test_support::assert_probe_default(
            &OhMyPiFactory,
            &[".omp", "agent", "sessions"],
        )
    }

    /// The slot is container framing: folded into the session losslessly, never
    /// a message, and the header behind it still drives every canonical field.
    #[tokio::test(flavor = "multi_thread")]
    async fn the_title_slot_is_folded_into_the_session() -> anyhow::Result<()> {
        let temp = TempDir::new()?;
        let store = ingest_fixtures(&temp).await?;

        let main = store.get_session(MAIN).await?.expect("main session lands");
        assert_eq!(main.session.source_agent, NAME);
        assert_eq!(&*main.session.project, PROJECT);
        let source = main.session.options.get("source").expect("source options");
        assert_eq!(source.get("adapter"), Some(&json!(NAME)));
        assert_eq!(source.get("format"), Some(&json!(3)));
        assert_eq!(
            source.get("title_slot").and_then(|slot| slot.get("title")),
            Some(&json!("harness problem")),
            "the stripped slot survives in options rather than being dropped",
        );
        // The placement hint is omp's hashed bucket, not a pi-style slug.
        assert_eq!(
            source.get("project_slug"),
            Some(&json!(BUCKET)),
            "the placement hint is omp's scope-encoded bucket, verbatim",
        );

        assert!(
            !main.messages.iter().any(|message| matches!(
                &message.message,
                Message::System { content, .. } if content.as_deref().map(String::as_str) == Some("title")
            )),
            "the slot must not mint a carrier message",
        );
        Ok(())
    }

    /// A legacy slot-less file is still what omp reads for older sessions, so it
    /// must read through the same path with no slot recorded.
    #[tokio::test(flavor = "multi_thread")]
    async fn a_legacy_slotless_session_ingests() -> anyhow::Result<()> {
        let temp = TempDir::new()?;
        let store = ingest_fixtures(&temp).await?;

        let legacy = store.get_session(LEGACY).await?.expect("legacy session");
        assert_eq!(&*legacy.session.project, PROJECT);
        assert!(
            legacy
                .session
                .options
                .get("source")
                .and_then(|source| source.get("title_slot"))
                .is_none(),
            "no slot in the file means no slot in options",
        );
        assert!(
            legacy
                .messages
                .iter()
                .any(|message| matches!(&message.message, Message::User { .. })),
            "the legacy transcript's user turn still lands",
        );
        Ok(())
    }

    /// omp's wider entry taxonomy rides the shared v3 carrier path: conversation
    /// stays conversation, orchestration stays System, nothing is dropped
    /// (spec.md#model-part-provenance placement rule 3).
    #[tokio::test(flavor = "multi_thread")]
    async fn omp_only_entry_types_become_system_carriers() -> anyhow::Result<()> {
        let temp = TempDir::new()?;
        let store = ingest_fixtures(&temp).await?;
        let main = store.get_session(MAIN).await?.expect("main session lands");

        let carriers: Vec<Option<String>> = main
            .messages
            .iter()
            .filter_map(|message| match &message.message {
                Message::System { content, .. } => Some(content.as_deref().cloned()),
                _ => None,
            })
            .collect();
        assert!(
            carriers.contains(&Some("ttsr_injection".to_owned())),
            "an omp-only entry type must survive as a carrier: {carriers:?}",
        );
        assert!(
            carriers.contains(&Some("model_change".to_owned())),
            "a shared pi carrier still maps: {carriers:?}",
        );

        let count = |want: fn(&Message) -> bool| {
            main.messages
                .iter()
                .filter(|message| want(&message.message))
                .count()
        };
        assert_eq!(count(|m| matches!(m, Message::User { .. })), 2);
        assert_eq!(count(|m| matches!(m, Message::Assistant { .. })), 1);
        assert_eq!(count(|m| matches!(m, Message::Tool { .. })), 1);

        let fork = store.get_session(FORK).await?.expect("fork session lands");
        assert!(
            fork.messages.iter().any(|message| matches!(
                &message.message,
                Message::System { content, .. } if content.as_deref().map(String::as_str) == Some("branch_summary")
            )),
            "omp's branch summary is orchestration, not conversation",
        );
        Ok(())
    }

    /// omp's `parentSession` is opaque (id OR path, by flow), so it is carried
    /// verbatim and never guessed into lineage (spec.md#model-no-synthesis).
    #[tokio::test(flavor = "multi_thread")]
    async fn the_opaque_parent_marker_is_carried_not_interpreted() -> anyhow::Result<()> {
        let temp = TempDir::new()?;
        let store = ingest_fixtures(&temp).await?;

        let fork = store.get_session(FORK).await?.expect("fork session lands");
        assert_eq!(fork.session.parent_session_id, None);
        assert_eq!(
            fork.session
                .options
                .get("source")
                .and_then(|source| source.get("raw_record"))
                .and_then(|raw| raw.get("parentSession")),
            Some(&json!(MAIN)),
        );
        Ok(())
    }

    /// The slot is framing only where omp writes it. A `title`-shaped row deeper
    /// in the file is not framing, so it must stay visible as a carrier instead
    /// of being skipped into nothing.
    #[test]
    fn a_title_shaped_row_off_line_one_stays_visible() {
        let adapter = OhMyPiAdapter::new(FIXTURES);
        let session = a_session();
        let slot = json!({"type": "title", "v": 1, "title": "t", "updatedAt": "x", "pad": " "});

        let framing = adapter
            .events_from_row(
                &session,
                &BoundedRow {
                    line: 1,
                    value: slot.clone(),
                },
                &mut (),
            )
            .expect("line 1 is framing");
        assert!(framing.is_empty(), "the physical slot yields no event");

        let deeper = adapter
            .events_from_row(
                &session,
                &BoundedRow {
                    line: 7,
                    value: slot,
                },
                &mut (),
            )
            .expect("a later row still maps");
        assert!(
            deeper
                .iter()
                .any(|event| matches!(event, IngestEvent::Message(Message::System { .. }))),
            "a title-shaped row off line 1 becomes a carrier, never a silent drop",
        );
    }

    /// The freshness gate must see the id behind the slot, or every sync
    /// re-reads the whole omp corpus.
    #[test]
    fn the_freshness_peek_reads_through_the_title_slot() {
        let path = main_file();
        let adapter = OhMyPiAdapter::new(FIXTURES);

        let first = peek_first_line(&path).expect("first line");
        assert_eq!(
            adapter.peek_session_id(&path, &first),
            Some(MAIN.to_owned()),
            "a slot-fronted file still gates on its session id",
        );
        assert!(matches!(
            adapter.peek_watermark(&path),
            SourceWatermark::At(_)
        ));
    }

    /// A file omp created but has not materialized yet (slot only) has no id and
    /// no watermark: `Opaque` re-reads rather than claiming freshness.
    #[test]
    fn a_slot_only_file_is_opaque() -> anyhow::Result<()> {
        let temp = TempDir::new()?;
        let bucket = temp.path().join("sessions").join("home-omp-x-1");
        std::fs::create_dir_all(&bucket)?;
        let path = bucket.join("2026-08-12T11-00-00-000Z_slotonly0000001.jsonl");
        let body = std::fs::read_to_string(main_file())?;
        let slot_line = body.lines().next().expect("slot line");
        std::fs::write(&path, format!("{slot_line}\n"))?;

        let adapter = OhMyPiAdapter::new(temp.path().join("sessions"));
        let first = peek_first_line(&path).expect("first line");
        assert_eq!(adapter.peek_session_id(&path, &first), None);
        assert!(matches!(
            adapter.peek_watermark(&path),
            SourceWatermark::Opaque
        ));
        Ok(())
    }

    /// A newer session version is a visible, counted skip that names the file and
    /// the fix - never a half-understood ingest.
    #[test]
    fn a_newer_session_version_is_a_named_skip() {
        let adapter = OhMyPiAdapter::new(FIXTURES);
        let path = Path::new("/tmp/omp/sessions/bucket/x.jsonl");
        let slot = json!({"type": "title", "v": 1, "title": "t", "updatedAt": "x", "pad": " "});
        let rows = |version: i64| {
            vec![
                BoundedRow {
                    line: 1,
                    value: slot.clone(),
                },
                BoundedRow {
                    line: 2,
                    value: json!({
                        "type": "session", "version": version, "id": "sess1",
                        "timestamp": "2026-08-12T10:00:00.000Z", "cwd": PROJECT,
                    }),
                },
            ]
        };

        let reason = adapter
            .unsupported_reason(path, &rows(4))
            .expect("a newer version is unsupported");
        assert!(reason.contains("version 4"), "{reason}");
        assert!(reason.contains("upgrade pond"), "{reason}");
        assert!(
            adapter.unsupported_reason(path, &rows(3)).is_none(),
            "the supported version reads normally",
        );
    }

    /// A moved CONTAINER must be as visible as a moved version. Without this the
    /// file falls through `session()` into `SkipReason::Empty`, which prints
    /// `empty` with no reason and shrinks the progress denominator - so a wholly
    /// unreadable corpus would report as a clean sync with nothing to do.
    #[test]
    fn a_moved_container_is_a_named_skip_not_an_empty_file() {
        let adapter = OhMyPiAdapter::new(FIXTURES);
        let path = Path::new("/tmp/omp/sessions/bucket/x.jsonl");
        let head = json!({
            "type": "session", "version": 3, "id": "sess1",
            "timestamp": "2026-08-12T10:00:00.000Z", "cwd": PROJECT,
        });

        // A slot shape this build does not know.
        let bumped_slot = vec![
            BoundedRow {
                line: 1,
                value: json!({"type": "title", "v": 2, "title": "t", "pad": " "}),
            },
            BoundedRow {
                line: 2,
                value: head.clone(),
            },
        ];
        let reason = adapter
            .unsupported_reason(path, &bumped_slot)
            .expect("a bumped slot version is unsupported");
        assert!(reason.contains("title-slot version 2"), "{reason}");
        assert!(reason.contains("upgrade pond"), "{reason}");

        // The harness-v2 container move pi already made, which this reader does
        // not decode - it must not read as an empty file.
        let v4_container = vec![BoundedRow {
            line: 1,
            value: json!({"kind": "header", "version": 4, "id": "sess1", "cwd": PROJECT}),
        }];
        let reason = adapter
            .unsupported_reason(path, &v4_container)
            .expect("a kind:header container is unsupported");
        assert!(reason.contains("version 4"), "{reason}");
        assert!(reason.contains("upgrade pond"), "{reason}");

        // And the shipped shape still reads normally.
        assert!(
            adapter
                .unsupported_reason(
                    path,
                    &[BoundedRow {
                        line: 1,
                        value: head
                    }]
                )
                .is_none(),
        );
    }

    /// Ingest-only: the refusal names the working alternative, per the CLI error
    /// contract.
    #[test]
    fn restore_refuses_and_names_the_alternative() {
        let session = crate::sessions::SessionWithMessages {
            session: a_session(),
            messages: Vec::new(),
        };

        // The CLI asks this BEFORE planning, so it is the surface that decides
        // the exit code; `serialize` only backs it up for a caller that skips it.
        let reason = OhMyPiFactory
            .restore_unsupported()
            .expect("oh-my-pi is ingest-only");
        assert!(reason.contains("pi-coding-agent"), "{reason}");
        assert!(reason.contains("--out-dir ~/.omp/agent"), "{reason}");

        let error = OhMyPiFactory
            .serialize(&session, RestoreFidelity::Native)
            .expect_err("serialize stays an error, not a panic");
        assert!(error.to_string().contains(reason), "both surfaces agree");
    }
}