import io, sys
ROOT = r"C:/Users/decid/Documents/projects/spt-core/.worktrees/351-w4/"
edits = {}

def ed(path, old, new, count=1):
    edits.setdefault(path, []).append((old, new, count))

SO = "crates/spt-daemon/src/shellout.rs"

# ── module doc :9-11 (stale "ships at once only when its queue is empty")
ed(SO, """//! - **One queue, one order.** A payload ships at once only when its queue is
//!   empty and an active instance is known. Otherwise it is appended and the
//!   queue drains oldest-first, stopping at the first payload that cannot ship.
""", """//! - **One queue, one order.** Every payload is appended first, reserving its
//!   place, and the queue drains oldest-first while an active instance is
//!   known, stopping at the first payload that cannot ship.
""")

# ── N1: the row answered NotLive leaves unconditionally
ed(SO, """                Ship::NotLive(why) => {
                    drop_not_live(dir, &mut report, &why)?;
                    if row.met_vacancy {
                        report.held = Some(why);
                        return Ok(report);
                    }
                    // This row was dropped with the rest; the pass goes on
                    // with a fresh read, since the queue just changed.
                    break;
                }
""", """                Ship::NotLive(why) => {
                    if row.met_vacancy {
                        drop_not_live(dir, &mut report, &why)?;
                        report.held = Some(why);
                        return Ok(report);
                    }
                    // The row the shipper answered for leaves the queue
                    // whatever its body parses as (gate fixup N1). Leaving it
                    // to drop_not_live's body test would ship it again on the
                    // next read, forever and under the drain lock, if the
                    // shipper and that test ever disagreed about it.
                    spool::delete_outbound_at(dir, row.id).map_err(store_err)?;
                    report.dropped.push(row.id);
                    drop_not_live(dir, &mut report, &why)?;
                    // The pass goes on with a fresh read, since the queue
                    // just changed.
                    break;
                }
""")
ed(SO, """/// **A not-live answer is about the owner, for this pass.** On it, every
/// pending live-only row that never met a vacancy is dropped at once, without
/// being shipped""", """/// **A not-live answer is about the owner, for this pass.** The row it
/// answered for is dropped, unless it met a vacancy. Every other pending
/// live-only row that never met a vacancy is dropped at once, without being
/// shipped""")

# ── N3: Delivered carries the owner it reached, never None
ed(SO, """    /// It reached the owner's active instance. `None` when another process's
    /// drain shipped it, where the instance it reached is not tracked.
    Delivered(Option<ActiveOwner>),
""", """    /// This caller's own drain shipped it to the owner's active instance,
    /// which it names. A row another drain shipped is [`Submitted::Handed`].
    Delivered(ActiveOwner),
""")
ed(SO, """    if report.shipped.contains(&id) {
        return Ok(Submitted::Delivered(report.reached));
    }
""", """    if report.shipped.contains(&id) {
        // `reached` is set with every ship; were it ever missing, the row is
        // gone and reads as HANDED below rather than claiming a destination.
        if let Some(to) = report.reached {
            return Ok(Submitted::Delivered(to));
        }
    }
""")

# ── N2: resolve AFTER the append, inside its transaction
ed(SO, """    // A payload submitted while the owner has no active instance MET A VACANCY
    // at its append (gate fixup R3-2): a drain pass that marked the queue just
    // before it was appended cannot have marked it.
    let vacant = resolve().is_none();
    let id = spool::spool_outbound_at(dir, from, body, kind, op_id).map_err(store_err)?;
    if vacant {
        spool::mark_outbound_row_vacancy_at(dir, id).map_err(store_err)?;
    }
""", """    // A payload appended while the owner has no active instance MET A VACANCY
    // (gate fixups R3-2 and N2). The owner is resolved AFTER the row is
    // written, inside the append's own transaction, so no drain pass can mark
    // the queue, or take this row, between that answer and the row carrying it.
    let id = spool::spool_outbound_at(dir, from, body, kind, op_id, &mut || resolve().is_none())
        .map_err(store_err)?;
""")

# test helpers
ed(SO, """        spool::spool_outbound_at(dir, "mock-0", body, KIND_TEXT, op).unwrap()
""", """        spool::spool_outbound_at(dir, "mock-0", body, KIND_TEXT, op, &mut || false).unwrap()
""")
ed(SO, """        spool::spool_outbound_at(dir, "mock-0", &frame, KIND_SENSORY, op).unwrap()
""", """        spool::spool_outbound_at(dir, "mock-0", &frame, KIND_SENSORY, op, &mut || false).unwrap()
""")
ed(SO, """        assert_eq!(out, Submitted::Delivered(Some(remote(FAR))));
""", """        assert_eq!(out, Submitted::Delivered(remote(FAR)));
""")
# stale fast-path wording in unit docs (:1053, :1092)
ed(SO, """    // [unit->REQ-SHELL-OUTBOUND-SPOOL] doyle gate fixup F2, condition 2 on the
    // FAST PATH: A holds the lock on an empty queue and is mid-ship when B
""", """    // [unit->REQ-SHELL-OUTBOUND-SPOOL] doyle gate fixup F2, condition 2 with an
    // EMPTY QUEUE: A holds the lock on an empty queue and is mid-ship when B
""")
ed(SO, """    // [unit->REQ-SHELL-OUTBOUND-SPOOL] doyle gate fixup F2's other half: a
    // fast-path row that DELIVERED is deleted, so the queue does not grow
""", """    // [unit->REQ-SHELL-OUTBOUND-SPOOL] doyle gate fixup F2's other half: a row
    // that never met a vacancy is deleted once DELIVERED, so the queue does not grow
""")

# new units, appended before the file's closing brace of the test module
ed(SO, """            assert_eq!(shipper.ship(&sensory_kind_text_body, &ActiveOwner::Local), Ship::Delivered);
            assert_eq!(spool::pending_count_at(&perch).unwrap(), 1, "it took the text legs");
        });
    }
}
""", """            assert_eq!(shipper.ship(&sensory_kind_text_body, &ActiveOwner::Local), Ship::Delivered);
            assert_eq!(spool::pending_count_at(&perch).unwrap(), 1, "it took the text legs");
        });
    }

    /// A shipper that answers NOT LIVE to everything, whatever the body.
    struct NotLiveToAll(Vec<String>);

    impl Shipper for NotLiveToAll {
        fn ship(&mut self, row: &OutboundRow, _to: &ActiveOwner) -> Ship {
            self.0.push(row.op_id.clone());
            Ship::NotLive("the owner is not live".into())
        }
    }

    // [unit->REQ-SHELL-OUTBOUND-SPOOL] doyle gate fixup N1: a NOT LIVE answer
    // for a row that never met a vacancy takes THAT row off the queue, even
    // when its body is text that drop_not_live's body test would keep. One
    // attempt, the row is gone, and the drain ends. Red-on-purpose: leaving
    // the answered row to the body test re-ships it forever.
    #[test]
    fn a_not_live_answer_for_a_text_row_drops_it_and_the_drain_ends() {
        let tmp = tempfile::tempdir().unwrap();
        let dir = tmp.path().join("q");
        let id = append(&dir, "words", "op-t");
        let mut to = || Some(remote(FAR));
        let mut rx = NotLiveToAll(Vec::new());
        let r = drain_queue(&dir, &mut to, &mut rx).unwrap();
        assert_eq!(rx.0, vec!["op-t"], "exactly one attempt");
        assert_eq!(r.dropped, vec![id]);
        assert!(spool::peek_outbound_at(&dir).unwrap().is_empty(), "the row left the queue");
    }

    // [unit->REQ-SHELL-OUTBOUND-SPOOL] doyle gate fixup R3-1, the text half: a
    // TEXT row that never met a vacancy, queued behind a held vacancy head,
    // STAYS when the owner answers not live. drop_not_live drops live-only
    // rows only. Red-on-purpose: dropping it by the vacancy mark alone loses
    // the text.
    #[test]
    fn an_unmarked_text_row_behind_a_held_vacancy_head_stays() {
        let tmp = tempfile::tempdir().unwrap();
        let dir = tmp.path().join("q");
        let head = sensory(&dir, "op-head");
        let mut none = || None;
        let mut rx = not_live();
        drain_queue(&dir, &mut none, &mut rx).unwrap();
        let text = append(&dir, "behind", "op-t");
        let rows = spool::peek_outbound_at(&dir).unwrap();
        assert!(rows[0].met_vacancy && !rows[1].met_vacancy, "precondition: {rows:?}");
        let mut to = || Some(remote(FAR));
        let r = drain_queue(&dir, &mut to, &mut rx).unwrap();
        assert!(r.held.is_some() && r.dropped.is_empty(), "{r:?}");
        let left: Vec<i64> = spool::peek_outbound_at(&dir).unwrap().iter().map(|r| r.id).collect();
        assert_eq!(left, vec![head, text], "the text row stays, behind the head");
    }

    // [unit->REQ-SHELL-OUTBOUND-SPOOL] doyle gate fixup N2: the vacancy answer
    // is taken INSIDE the append. A row appended while the owner has no
    // active instance carries the mark from its first read; one appended
    // while it has one does not.
    #[test]
    fn the_append_carries_the_vacancy_answer_it_was_written_under() {
        let tmp = tempfile::tempdir().unwrap();
        let dir = tmp.path().join("q");
        spool::spool_outbound_at(&dir, "mock-0", "v", KIND_TEXT, "op-v", &mut || true).unwrap();
        spool::spool_outbound_at(&dir, "mock-0", "a", KIND_TEXT, "op-a", &mut || false).unwrap();
        let marks: Vec<bool> =
            spool::peek_outbound_at(&dir).unwrap().iter().map(|r| r.met_vacancy).collect();
        assert_eq!(marks, vec![true, false]);
    }
}
""")

# ── spool.rs
SP = "crates/spt-store/src/spool.rs"
ed(SP, """/// retry of it ships under. Returns the row id, which is its place in the
/// send order. No `.has-messages` sentinel: nothing listens on a sender's
/// queue.
// [impl->REQ-SHELL-OUTBOUND-SPOOL]
pub fn spool_outbound_at(
    dir: &Path,
    from_id: &str,
    body: &str,
    kind: &str,
    op_id: &str,
) -> rusqlite::Result<i64> {
    let conn = open_spool_at(dir)?;
    conn.execute_batch("BEGIN IMMEDIATE")?;
    insert_message(&conn, from_id, body, WINDOW_DEFAULT, kind, false)?;
    let id = conn.last_insert_rowid();
    conn.execute("UPDATE messages SET op_id = ?1 WHERE id = ?2", params![op_id, id])?;
    conn.execute_batch("COMMIT")?;
    Ok(id)
}
""", """/// retry of it ships under. Returns the row id, which is its place in the
/// send order. No `.has-messages` sentinel: nothing listens on a sender's
/// queue.
///
/// **`vacant` is asked AFTER the row is written, inside the same
/// transaction** (gate fixup N2). `true` means the owner has no active
/// instance, and the row commits already marked as having MET A VACANCY. The
/// write lock is held across the question, so no drain pass can mark the
/// queue, or take the row, between the answer and the row carrying it.
// [impl->REQ-SHELL-OUTBOUND-SPOOL]
pub fn spool_outbound_at(
    dir: &Path,
    from_id: &str,
    body: &str,
    kind: &str,
    op_id: &str,
    vacant: &mut dyn FnMut() -> bool,
) -> rusqlite::Result<i64> {
    let conn = open_spool_at(dir)?;
    conn.execute_batch("BEGIN IMMEDIATE")?;
    insert_message(&conn, from_id, body, WINDOW_DEFAULT, kind, false)?;
    let id = conn.last_insert_rowid();
    let met_vacancy = i64::from(vacant());
    conn.execute(
        "UPDATE messages SET op_id = ?1, met_vacancy = ?2 WHERE id = ?3",
        params![op_id, met_vacancy, id],
    )?;
    conn.execute_batch("COMMIT")?;
    Ok(id)
}
""")
ed(SP, """/// Record that SENDER-side row `id` met a vacancy at its APPEND: the owner had
/// no active instance when it was submitted (gate fixup R3-2). A drain pass
/// that marked the queue a moment earlier cannot have marked it.
// [impl->REQ-SHELL-OUTBOUND-SPOOL]
pub fn mark_outbound_row_vacancy_at(dir: &Path, id: i64) -> rusqlite::Result<()> {
    let conn = open_spool_at(dir)?;
    conn.execute("UPDATE messages SET met_vacancy = 1 WHERE id = ?1", params![id])?;
    Ok(())
}

""", "")

# ── spt-store shellout.rs tests
SS = "crates/spt-store/src/shellout.rs"
ed(SS, """spool_outbound_at(&q, &id, "hello", "text", "op-1").unwrap()""",
   """spool_outbound_at(&q, &id, "hello", "text", "op-1", &mut || false).unwrap()""")
ed(SS, """spool_outbound_at(&q, "mock-0", "warm", "shell-sensory", "warm").unwrap()""",
   """spool_outbound_at(&q, "mock-0", "warm", "shell-sensory", "warm", &mut || false).unwrap()""")
ed(SS, """spool_outbound_at(&q, "mock-0", frame, "shell-sensory", &format!("{op}{i}")).unwrap()""",
   """spool_outbound_at(&q, "mock-0", frame, "shell-sensory", &format!("{op}{i}"), &mut || false)
                .unwrap()""")

# ── reporting.rs
RP = "crates/spt/src/api/reporting.rs"
ed(RP, """        Ok(Submitted::Delivered(Some(ActiveOwner::Local))) => 0,
        Ok(Submitted::Delivered(Some(ActiveOwner::Remote { node, .. }))) => {
            spt_proto::emit_line_err!(
                "SENSORY_DELIVERED:{shell_id} -> {owner}@{node} type={sensory_type}"
            );
            0
        }
        Ok(Submitted::Delivered(None)) => {
            spt_proto::emit_line_err!("SENSORY_DELIVERED:{shell_id} -> {owner} type={sensory_type}");
            0
        }
""", """        Ok(Submitted::Delivered(ActiveOwner::Local)) => 0,
        Ok(Submitted::Delivered(ActiveOwner::Remote { node, .. })) => {
            spt_proto::emit_line_err!(
                "SENSORY_DELIVERED:{shell_id} -> {owner}@{node} type={sensory_type}"
            );
            0
        }
""")

# ── cli.rs
CL = "crates/spt/src/cli.rs"
ed(CL, """        Ok(Submitted::Delivered(to)) => {
            if let Some(ActiveOwner::Remote { node, .. }) = to {
""", """        Ok(Submitted::Delivered(to)) => {
            if let ActiveOwner::Remote { node, .. } = to {
""")
ed(CL, """        assert!(shell_text_refusal("hello").is_none());
        assert!(shell_text_refusal("I saw type=sensory in a log").is_none());
    }
""", """        assert!(shell_text_refusal("hello").is_none());
        assert!(shell_text_refusal("I saw type=sensory in a log").is_none());
    }

    // [unit->REQ-SHELLS-FOLLOW-ACTIVE] doyle gate fixup R3-4 at the VERB: a
    // `shell-say` send (the proven shell extra) whose body is a sensory EVENT
    // is refused through `cmd_send_verdict`, and nothing reaches the shell's
    // queue. The owner has no active instance, so a body that got past the
    // refusal would queue and read as delivered. Control: plain text does.
    // Red-on-purpose: deleting the refusal call in `finish_shell_send` queues
    // the frame.
    #[test]
    fn a_shell_say_verb_refuses_a_sensory_event_body() {
        let home = crate::testutil::isolated_home();
        let _ = &home;
        let _env = NoSessionEnv::new();
        let send = |body: &str| {
            cmd_send_verdict(
                "ling".into(),
                None,
                false,
                false,
                false,
                false,
                false,
                None,
                false,
                None,
                false,
                Some(body.to_string()),
                SendExtras {
                    shell: Some("mock-0".into()),
                    ..Default::default()
                },
            )
        };
        let owlery = perch::owlery_dir();
        let frame = spt_daemon::shellchan::compose_sensory_frame("mock-0", "bumped", "w", 1);
        assert_eq!(send(&frame), SendVerdict::Refused { code: 1 });
        assert!(!spt_daemon::shellout::has_queued(&owlery, "ling"), "nothing was queued");
        assert_eq!(send("plain words"), SendVerdict::Delivered, "control: text passes");
        assert!(spt_daemon::shellout::has_queued(&owlery, "ling"), "control: it queued in the vacancy");
    }
""")

# ── twohost_axes.rs
TA = "crates/spt-daemon/tests/twohost_axes.rs"
ed(TA, """            Submitted::Delivered(Some(ActiveOwner::Remote {
                node: winner.clone(),
                uid: Some(AXES_UID.to_string()),
            })),
""", """            Submitted::Delivered(ActiveOwner::Remote {
                node: winner.clone(),
                uid: Some(AXES_UID.to_string()),
            }),
""")
ed(TA, """Submitted::Delivered(Some(ActiveOwner::Remote { .. }))""",
   """Submitted::Delivered(ActiveOwner::Remote { .. })""")

ok = True
out = {}
for path, lst in edits.items():
    full = ROOT + path
    with open(full, 'r', encoding='utf-8', newline='') as f:
        s = f.read()
    for old, new, count in lst:
        o = old.replace(chr(10), chr(13)+chr(10)); n = new.replace(chr(10), chr(13)+chr(10))
        c = s.count(o)
        if c != count:
            print(f"FAIL {path}: expected {count} got {c}: {old[:70]!r}"); ok = False; continue
        s = s.replace(o, n)
    out[full] = s
if not ok:
    print("NOTHING WRITTEN"); sys.exit(1)
for full, s in out.items():
    with open(full, 'w', encoding='utf-8', newline='') as f:
        f.write(s)
    print("wrote", full)