myxce/src/scheduler/testsuite/test_repl_prompt_context.c

/*
 * test_repl_prompt_context.c - contract test for the interactive simulation
 * REPL's Prompt Context component (scheduler-sim-repl-verbosity spec, task
 * 6.1, Requirement 5.5): the readline() prompt shown by repl_command_loop()
 * (repl.c) is the plain "(sim) " before any session has completed, and
 * becomes "(sim session=<N> t=<ISO8601-ts>) " (bare rs->session_count, the
 * existing repl_format_ts_human() full-ISO8601 form) once at least one
 * session has run -- in both `step` and `continue` modes -- without
 * altering any command's own printed output.
 *
 * RED-phase-only (task 6.1): this file does NOT implement the dynamic
 * prompt (task 6.2, a separate later dispatch) -- today's XCE_REPL_PROMPT
 * macro (repl.c) is a static "(sim) " literal that never varies, so the
 * session/timestamp-bearing assertions below are expected to FAIL against
 * the current implementation. The "before any session" baseline test is
 * expected to PASS today (today's prompt IS always "(sim) ") -- it is
 * task 6.2's own regression-safety net, not new RED evidence.
 *
 * Split into its own file rather than appended to
 * test_repl_execution_control.c (design.md's original File Structure Plan
 * target): that file is 948 lines today -- already over this project's
 * 800-line cap, and grown 11 lines past the 937-line figure design.md's
 * own text cited when it named that file as this task's home. Landing this
 * task's new PTY-based capture harness (openpty()-based, materially
 * different plumbing from every existing helper in that file) plus 2 test
 * functions there would push it toward/past the ~1000-line threshold this
 * exact spec family already treats as a split trigger (see that file's own
 * task-2.1/3.1 "Correction" notes, and test_repl_round_rollup.c/
 * test_repl_event_echo.c's own meson.build comments for the established
 * precedent of splitting rather than growing an already-over-cap sibling).
 *
 * Why a PTY, not this file family's usual freopen()-from-a-script-file
 * capture technique: readline()/libedit only ever WRITES a prompt string
 * to its output stream when BOTH its input and output are connected to a
 * real terminal. Empirically verified (outside this test binary, via a
 * throwaway libedit probe program) before writing this file:
 *   - stdin from a regular file, stdout to a regular file (this file
 *     family's existing freopen() technique, used by every sibling test in
 *     test_repl_execution_control.c/test_repl_round_rollup.c/
 *     test_repl_event_echo.c) -- readline() reads lines correctly but NEVER
 *     writes the prompt text at all.
 *   - stdin from a file, stdout to a pty -- still no prompt written.
 *   - stdin from a pty, stdout to a regular file -- still no prompt written.
 *   - BOTH stdin and stdout connected to the SAME pty (a genuinely
 *     interactive session) -- and only then -- the prompt is written, on
 *     every readline() call.
 * So this file's own run_repl_via_pty() below opens a pty pair (openpty(),
 * <pty.h>) and dup2()s the slave onto the child's stdin/stdout/stderr
 * (instead of freopen()-ing from/to plain files), matching the ONE
 * configuration that actually makes the prompt observable. ECHO is
 * disabled on the slave (termios, <termios.h>) before the child starts --
 * verified separately that this does NOT suppress readline()'s own prompt
 * output (which is unconditional whenever both ends are a tty), it only
 * suppresses the pty line discipline's redundant re-echo of each typed
 * character, which would otherwise interleave with and obscure the actual
 * program output this file's assertions care about.
 *
 * White-box #include set: identical to test_repl_execution_control.c's own
 * (see that file's header comment for the full per-TU rationale) -- this
 * file exercises the same real sched_sim_driver_run()/repl_command_loop()
 * path, just through a different stdin/stdout plumbing technique.
 */
#include <ctype.h>
#include <errno.h>
#include <limits.h>
#include <pty.h>
#include <stdatomic.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/wait.h>
#include <termios.h>
#include <unistd.h>

#include <unity/unity.h>

#include "controller.capnp.h"
#include "kvmgr.h"
#include "scheduler.h"
#include "xstring.h"

/* Pull in the real functions under test into this TU (white-box #include,
 * same convention every sibling in this file family already establishes). */
#include "../src/core.c"
#include "../src/core_clock.c"
#include "../src/core_events.c"
#include "../src/core_session.c"
#include "../src/decision.c"
#include "../src/plugin.c"
#include "../src/plugins/builtin_fcfs.c"
#include "../src/recording.c"
#include "../src/recording_session.c"
#include "../src/repl.c"
#include "../src/repl_fmt.c"
#include "../src/repl_rollup.c"
#include "../src/repl_break.c"
#include "../src/repl_info.c"
#include "../src/repl_params.c"
#include "../src/sim.c"
#include "../src/sim_loop.c"
#include "../src/sim_picker.c"
#include "../src/sim_limit.c" /* bug-1825: sched_sim_driver_run() now also calls sim_bootstrap_limit_plugin() (sim_loop.c-cap sibling, sim_picker.c's own bug-1567 precedent) as an extern -- undefined reference at link time otherwise */
#include "../src/sim_record.c"
#include "../src/sim_verify.c"
#include "../src/snapshot.c"
#include "../src/snapshot_load.c"
#include "../src/snapshot_wire.c"

void kill_sched_job(void* arg) { (void)arg; }
void kill_sched_queue(void* arg) { (void)arg; }
void kill_sched_worker(void* arg) { (void)arg; }
void kill_sched_user(void* arg) { (void)arg; }
void kill_sched_project(void* arg) { (void)arg; }
void kill_sched_resval(xce_val_t* val) { (void)val; }

/* repl_info.c (scheduler-sim-diagnostics spec, task 2.1) now calls
 * limit_get_configured_limits() (plugins/limit.c, not part of this TU) to
 * resolve `info limit`'s name/all argument -- link-satisfying stub only,
 * same rationale as kill_sched_job()/kill_sched_resval() above; this
 * file's own fixtures never register the limit plugin, so NULL
 * (limit_get_configured_limits()'s own documented "plugin absent" return)
 * is also the behaviorally correct answer here, not just a link
 * placeholder. repl_info_limit() itself is not yet wired into
 * repl_cmd_info() (task 4), so no test in this file can reach it anyway. */
var limit_get_configured_limits(xce_sched_core_t* core)
{
    (void)core;
    return NULL;
}

/* scheduler-sim-diagnostics spec, task 3.1: repl_info.c now also calls
 * limit_get_usage_instances() (plugins/limit.c, not part of this TU) to
 * read a limit's live usage instances -- identical link-satisfying stub,
 * same rationale as limit_get_configured_limits() immediately above; this
 * file's own fixtures never register the limit plugin, so NULL
 * (limit_get_usage_instances()'s own documented "plugin absent" return) is
 * also the behaviorally correct answer here, not just a link placeholder. */
var limit_get_usage_instances(xce_sched_core_t* core)
{
    (void)core;
    return NULL;
}

/* scheduler-sim-diagnostics spec, task 6 (bug-1825): sim_loop.c's
 * sched_sim_driver_run() now also calls sim_bootstrap_limit_plugin()
 * (sim_limit.c, pulled into this TU via the #include above), which in
 * turn calls limit_new_plugin() (plugins/limit.c, not part of this TU) --
 * identical link-satisfying stub rationale as limit_get_configured_
 * limits()/limit_get_usage_instances() immediately above. Returning NULL
 * here is also the behaviorally correct answer, not just a link
 * placeholder: sim_bootstrap_limit_plugin() already handles a NULL
 * limit_new_plugin() return gracefully (logs an ERROR, registers nothing,
 * returns) -- the exact same "out of memory" path sim_bootstrap_picker()
 * (sim_picker.c) takes for a NULL builtin_fcfs_new_plugin() -- so this
 * file's fixtures simply end up with no limit plugin registered, same
 * observable state as before this task. */
xce_sched_plugin_t* limit_new_plugin(void)
{
    return NULL;
}

/* limit_seed_config() (plugins/limit.c, not part of this TU) -- bug-1825
 * remediation round 2's new honestly-typed config-seed entry point.
 * sim_bootstrap_limit_plugin() (sim_limit.c) only calls it after
 * plugin_register_core() has already succeeded against a real, non-NULL
 * limit_new_plugin() plugin -- unreachable in this file since the
 * limit_new_plugin() stub immediately above always returns NULL, but the
 * symbol must still resolve at link time. Same link-satisfying-stub
 * rationale as limit_new_plugin() immediately above. */
int limit_seed_config(xce_sched_plugin_t* plugin, xce_sched_core_t* core,
                      xce_conf4sched_t* config)
{
    (void)plugin;
    (void)core;
    (void)config;
    return 0;
}

/* Link-satisfying stubs -- same rationale as every sibling in this file
 * family (see test_repl_execution_control.c's own copies for the full
 * per-stub doc comments). */
int sched_reconcile_job_resource_usage(xce_sched_core_t*  c,
                                       xce_job_account_t* a,
                                       xce_job_rusages_t* r)
{
    (void)c; (void)a; (void)r; return 0;
}

void sched_worker_release_host_ctx(xce_sched_worker_t* w) { (void)w; }

int sched_fill_limit_requests(xce_sched_alloc_req_t* alloc_req,
                              xlist_t* limits, xlist_t* rusages, int nhosts,
                              int nslots)
{
    (void)alloc_req;
    (void)limits;
    (void)rusages;
    (void)nhosts;
    (void)nslots;
    return 0;
}

int sched_account_jobs(xce_sched_core_t* core)
{
    (void)core;
    return 0;
}

/* bug-1836 (scheduler-driver-equivalence fix, Recorded-Decision
 * Reconciliation capacity gate): link-satisfying stubs, mirroring this
 * file's existing sched_account_jobs() no-op precedent immediately above --
 * sim_loop.c's sim_reconcile_unaccounted_jobs() now also calls
 * sched_check_host_satisfy()/sched_check_shared_satisfy() (resource.c) to
 * re-validate a recorded decision against live capacity, and
 * sched_charge_account() (resource.c) to keep that check's own view of
 * capacity up to date across multiple candidates in the same round. This
 * file never exercises real resource contention (same rationale as the
 * existing sched_account_jobs()/sched_ctx_get_uncomit_host() no-ops), so
 * the satisfy checks trivially always succeed and the charge call is a
 * no-op. */
int sched_check_host_satisfy(xce_sched_core_t* core, xce_sched_worker_t* w,
                             xvec_t* uncomitted, xvec_idval_t* wanted,
                             xvec_idval_t* unmeet)
{
    (void)core;
    (void)w;
    (void)uncomitted;
    (void)wanted;
    (void)unmeet;
    return 1;
}

int sched_check_shared_satisfy(xce_sched_core_t* core, xvec_t* uncomitted,
                               xvec_idval_t* wanted, xvec_idval_t* unmeet)
{
    (void)core;
    (void)uncomitted;
    (void)wanted;
    (void)unmeet;
    return 1;
}

int sched_charge_account(xce_sched_core_t* core, xce_job_account_t* acct)
{
    (void)core;
    (void)acct;
    return 0;
}

int sched_uncharge_account(xce_sched_core_t* core, xce_job_account_t* acct)
{
    (void)core;
    (void)acct;
    return 0;
}


int sched_resolve_job_picker(xce_sched_core_t* core, xce_sched_plugin_t** out)
{
    xce_sched_plugin_t* plugin = NULL;
    int                 rc     = 0;

    if ((core == NULL) || (out == NULL)) {
        return -EINVAL;
    }

    *out = NULL;

    if (core->job_pick_plugin == NULL) {
        return -EINVAL;
    }

    lock(core->plugins, RTTI_LOCK_READ);
    rc = key_find(core->plugins, core->job_pick_plugin, (void**)&plugin);
    unlock(core->plugins);

    if (rc <= 0) {
        return (rc < 0) ? rc : -ENOENT;
    }

    if ((plugin == NULL) || (plugin->picker_iface == NULL)
        || (plugin->picker_iface->next_job == NULL)
        || (plugin->picker_iface->begin_session == NULL)
        || (plugin->picker_iface->end_session == NULL)
        || (plugin->picker_iface->on_dispatch == NULL)
        || (plugin->picker_iface->on_pend == NULL)) {
        return -EINVAL;
    }

    *out = plugin;

    return 0;
}

/* scheduler-sim-live-allocation spec, task 4.1: sim_loop.c's real
 * sched_sim_driver_run() (white-box #include'd into this TU above) now
 * spawns sched_job_picker()/sched_allocator() (sched_election.c) directly
 * instead of its own, now-removed sim_picker_thread_fn() -- an undefined
 * reference at link time in this white-box TU otherwise, since
 * sched_election.c's own sched_allocator() calls the REAL
 * sched_make_allocation()/sched_commit_decision() (scheduler.c), neither
 * part of this file's own hand-picked white-box #include set (this file's
 * own header comment lists that set). This file's own test(s) do not
 * exercise genuine resource-matching allocation, so this link-satisfying
 * pair faithfully reproduces the exact PEND-only contract the REMOVED
 * sim_picker_thread_fn() itself used to provide -- split across the SAME
 * two-thread jbuf hand-off the real production pair now uses (one
 * producer, one consumer, matching the project's jbuf capacity-1
 * rendezvous invariant) instead of the old single fused thread -- so this
 * file's own existing assertions are unaffected by this task's production
 * change. Genuine resource-based dispatch through the real, unmodified
 * sched_job_picker()/sched_allocator() pair is proved separately by
 * test_sim_live_allocation.c (task 4.1's own dedicated fixture, which
 * links the real scheduler_srcs instead of white-boxing individual files,
 * precisely so it can reach the real allocation chain this stub
 * deliberately does not). */
void* sched_job_picker(void* arg)
{
    xce_sched_core_t*   core   = (xce_sched_core_t*)arg;
    xce_sched_plugin_t* picker = NULL;

    for (;;) {
        int      go    = 0;
        uint64_t jobid = 0;

        cbuf_get(&(core->wbuf), &go);

        if (sched_resolve_job_picker(core, &picker) < 0) {
            jobid = 0;
            cbuf_put(&(core->jbuf), &jobid);
            continue;
        }

        picker->picker_iface->begin_session(picker);

        for (;;) {
            xce_sched_job_t* job = picker->picker_iface->next_job(picker);

            if (job == NULL) {
                break;
            }

            jobid = job->jobid;
            cbuf_put(&(core->jbuf), &jobid);
        }

        jobid = 0;
        cbuf_put(&(core->jbuf), &jobid);

        picker->picker_iface->end_session(picker);
    }

    return NULL;
}

void* sched_allocator(void* arg)
{
    xce_sched_core_t*   core   = (xce_sched_core_t*)arg;
    xce_sched_plugin_t* picker = NULL;

    for (;;) {
        uint64_t         jobid    = 0;
        xce_decision_t*  decision = NULL;
        xce_sched_job_t* job;

        cbuf_get(&(core->jbuf), &jobid);
        if (jobid == 0) {
            decision = NULL;
            cbuf_put(&(core->decisions), &decision);
            continue;
        }

        job = sched_get_job(core, jobid);
        if (job == NULL) {
            continue;
        }

        (void)sched_resolve_job_picker(core, &picker);
        if (picker != NULL) {
            picker->picker_iface->on_pend(picker, job->jobid);
        }

        decision = make_pend_decision(make_pendreason(job->jobid));
        cbuf_put(&(core->decisions), &decision);
    }

    return NULL;
}

/* scheduler-sim-live-allocation spec, task 4.1 (bug-1829 fix): sim_loop.c's
 * real sched_sim_driver_run() (white-box #include'd into this TU above) now
 * also calls sim_bootstrap_seqjob_allocator()/sim_bootstrap_parjob_
 * allocator() (sim_alloc.c) -- undefined references at link time in this
 * white-box TU otherwise, since neither symbol is part of this file's own
 * hand-picked white-box #include set. This file's own test(s) exercise the
 * link-satisfying sched_job_picker()/sched_allocator() stub pair above
 * (which never reaches real primary allocation), so a true no-op is the
 * correct, behavior-preserving stand-in here -- the real bootstrap logic is
 * exercised separately by test_sim_live_allocation.c (task 4.1's own
 * dedicated fixture, which links the real scheduler_srcs). */
void sim_bootstrap_seqjob_allocator(xce_sched_sim_driver_t* sim)
{
    (void)sim;
}

void sim_bootstrap_parjob_allocator(xce_sched_sim_driver_t* sim)
{
    (void)sim;
}

void setUp(void) { return; }
void tearDown(void) { return; }

/* ---------------------------------------------------------------------- */
/* Fixture helpers -- same techniques test_repl_execution_control.c/       */
/* test_repl_round_rollup.c already establish, duplicated per this file    */
/* family's own established convention (see e.g. test_repl_round_rollup.c's */
/* header comment) rather than factored into a shared header.              */
/* ---------------------------------------------------------------------- */

static void make_tmp_dir(char* out, size_t out_len, const char* suffix)
{
    char tmpl[PATH_MAX];

    snprintf(tmpl, sizeof(tmpl), "/tmp/sched_replprompt%s_XXXXXX", suffix);
    TEST_ASSERT_NOT_NULL(mkdtemp(tmpl));
    TEST_ASSERT_TRUE(strlen(tmpl) < out_len);
    strcpy(out, tmpl);
}

static void build_fixture_scheduler(xce_scheduler_t* sched, char* record_dir)
{
    int rc;

    memset(sched, 0, sizeof(*sched));
    mylock_init(&(sched->record_lock));

    rc = xce_sched_core_init(&(sched->core));
    TEST_ASSERT_EQUAL_INT(0, rc);

    sched->core.job_pick_plugin = (char*)"builtin_fcfs";
    sched->record_dir           = record_dir;
}

static FILE* write_header_and_snapshot(char* out_path, size_t out_path_len,
                                       const char* suffix)
{
    char            record_dir[PATH_MAX];
    xce_scheduler_t sched;
    int             rc;

    make_tmp_dir(record_dir, sizeof(record_dir), suffix);
    build_fixture_scheduler(&sched, record_dir);

    rc = sched_record_start(&sched, "fixture.replay");
    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_TRUE(strlen(sched.record_file_path) < out_path_len);
    strcpy(out_path, sched.record_file_path);

    return sched.record_file;
}

static void append_queue_update_event(FILE* f, uint64_t ts,
                                      const char* queue_name)
{
    xce_queue_t          queue;
    replay_event_t       src;
    struct capn          c;
    struct capn_segment* cs;
    ReplayEvent_ptr      p;

    memset(&queue, 0, sizeof(queue));
    queue.queue    = (char*)queue_name;
    queue.status   = QUEUE_STAT_OPEN;
    queue.priority = 1;

    memset(&src, 0, sizeof(src));
    src.ts                = ts;
    src.body_which        = ReplayEvent_body_queueUpdate;
    src.body.queue_update = &queue;

    capn_init_malloc(&c);
    cs = capn_root(&c).seg;
    encode_ReplayEvent_ptr(cs, &p, &src);
    capn_setp(capn_root(&c), 0, p.p);

    TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c));
    capn_free(&c);
}

static void append_jobs_push_event(FILE* f, uint64_t ts, uint64_t jobid,
                                   const char* queue_name)
{
    job_submit_data_t    submit;
    xce_job_t            job;
    xce_job_t*           jobs[1];
    xce_jobs_data_t      jobs_data;
    replay_event_t       src;
    struct capn          c;
    struct capn_segment* cs;
    ReplayEvent_ptr      p;

    memset(&submit, 0, sizeof(submit));
    submit.queue    = (char*)queue_name;
    submit.user     = (char*)"alice";
    submit.mintasks = 1;
    submit.maxtasks = 1;

    memset(&job, 0, sizeof(job));
    job.jobid  = jobid;
    job.state  = JOB_STAT_PEND;
    job.submit = &submit;

    jobs[0] = &job;

    memset(&jobs_data, 0, sizeof(jobs_data));
    jobs_data.n_total = 1;
    jobs_data.count   = 1;
    jobs_data.jobs    = jobs;

    memset(&src, 0, sizeof(src));
    src.ts             = ts;
    src.body_which     = ReplayEvent_body_jobsPush;
    src.body.jobs_push = &jobs_data;

    capn_init_malloc(&c);
    cs = capn_root(&c).seg;
    encode_ReplayEvent_ptr(cs, &p, &src);
    capn_setp(capn_root(&c), 0, p.p);

    TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c));
    capn_free(&c);
}

static void append_replay_session(FILE* f, uint64_t begin_round)
{
    struct capn          c;
    struct capn_segment* cs;
    ReplaySession_ptr    sp;

    capn_init_malloc(&c);
    cs = capn_root(&c).seg;
    sp = new_ReplaySession(cs);
    ReplaySession_set_beginRound(sp, begin_round);
    capn_setp(capn_root(&c), 0, sp.p);

    TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c));
    capn_free(&c);
}

static void append_end_of_stream_marker(FILE* f)
{
    replay_event_t       src;
    struct capn          c;
    struct capn_segment* cs;
    ReplayEvent_ptr      p;

    memset(&src, 0, sizeof(src));
    src.body_which = ReplayEvent_body_endSession;

    capn_init_malloc(&c);
    cs = capn_root(&c).seg;
    encode_ReplayEvent_ptr(cs, &p, &src);
    capn_setp(capn_root(&c), 0, p.p);

    TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c));
    capn_free(&c);
}

/* ---------------------------------------------------------------------- */
/* PTY capture harness -- see this file's own header comment for why a    */
/* pty (not the freopen()-from-a-file technique every sibling test file   */
/* uses) is required to observe readline()'s prompt text at all.          */
/* ---------------------------------------------------------------------- */

/*
 * run_repl_via_pty - forks a child whose stdin/stdout/stderr are all
 * dup2()'d onto ONE end (the slave) of a fresh pty pair, runs a real
 * sched_sim_driver_run() against `replay_path` there, writes `script`
 * (newline-terminated commands, same convention this file family's
 * write_text_file()-plus-freopen() scripts already use) into the master
 * end, then drains the master until the child closes its end (normal
 * completion via `quit`'s exit(0)) or a read error occurs -- Linux ptys
 * report EOF-after-last-close as EIO, not a 0-byte read, so this loop
 * treats ANY read() <= 0 as "done", not just 0.
 *
 * ECHO is disabled on the slave before the child starts (termios) so the
 * pty line discipline does not itself re-echo each byte written into the
 * master -- readline()'s OWN prompt/redraw behavior is independent of this
 * flag and is unaffected (verified: disabling it does not suppress the
 * prompt, only the kernel-level character echo that would otherwise
 * duplicate/obscure it in the captured buffer).
 *
 * Always returns a NUL-terminated, malloc'd buffer (never NULL) and
 * asserts the child exited via `quit`'s real exit(0) -- same "quit's
 * exit(0) is the only correct exit path" discipline every forked-child
 * test in this file family already enforces.
 */
static char* run_repl_via_pty(const char* replay_path, const char* script)
{
    int             master_fd;
    int             slave_fd;
    struct termios  tio;
    pid_t           pid;
    size_t          cap;
    size_t          len;
    char*           captured;
    int             status;

    TEST_ASSERT_EQUAL_INT_MESSAGE(
        0, openpty(&master_fd, &slave_fd, NULL, NULL, NULL),
        "openpty() must succeed -- this harness requires a real pty pair "
        "since readline() only writes its prompt when both ends are a tty");

    if (tcgetattr(slave_fd, &tio) == 0) {
        tio.c_lflag &= (tcflag_t)~ECHO;
        TEST_ASSERT_EQUAL_INT(0, tcsetattr(slave_fd, TCSANOW, &tio));
    }

    pid = fork();
    TEST_ASSERT_TRUE_MESSAGE(pid >= 0, "fork() must succeed");

    if (pid == 0) {
        xce_sched_sim_driver_t sim;
        sched_sim_args_t       args;

        close(master_fd);
        dup2(slave_fd, STDIN_FILENO);
        dup2(slave_fd, STDOUT_FILENO);
        dup2(slave_fd, STDERR_FILENO);
        if (slave_fd > STDERR_FILENO) {
            close(slave_fd);
        }

        memset(&sim, 0, sizeof(sim));
        memset(&args, 0, sizeof(args));
        args.replay_path  = replay_path;
        args.sim_start_ts = SCHED_SIM_START_TS_LATEST;
        args.sim_speed    = 0;
        args.interactive  = true;

        (void)sched_sim_driver_run(&sim, &args);

        /* Unreachable in a correct implementation -- the script's trailing
         * `quit` must already have called exit(0). */
        fflush(stdout);
        _exit(111);
    }

    close(slave_fd);

    TEST_ASSERT_TRUE_MESSAGE(
        write(master_fd, script, strlen(script)) == (ssize_t)strlen(script),
        "writing the scripted commands into the pty master must succeed");

    cap      = 4096;
    len      = 0;
    captured = (char*)malloc(cap);
    TEST_ASSERT_NOT_NULL(captured);

    for (;;) {
        ssize_t r;

        if (len + 1024 >= cap) {
            cap     *= 2;
            captured = (char*)realloc(captured, cap);
            TEST_ASSERT_NOT_NULL(captured);
        }

        r = read(master_fd, captured + len, cap - len - 1);
        if (r <= 0) {
            /* 0 == normal EOF; a negative read (Linux ptys report EIO once
             * the slave's last fd closes) is ALSO end-of-output here, not
             * an error -- both mean "the child is done writing". */
            break;
        }
        len += (size_t)r;
    }
    captured[len] = '\0';

    TEST_ASSERT_EQUAL_INT_MESSAGE(pid, waitpid(pid, &status, 0),
                                  "waitpid must reap the REPL child");
    TEST_ASSERT_TRUE_MESSAGE(WIFEXITED(status),
                             "the REPL child must exit normally via quit's "
                             "exit(0), not crash/signal");
    TEST_ASSERT_EQUAL_INT_MESSAGE(0, WEXITSTATUS(status),
                                  "quit's exit(0) is the only correct exit "
                                  "path");

    close(master_fd);

    return captured;
}

/*
 * looks_like_iso8601_ts - checks ONLY the shape of `s`'s first 20 bytes
 * against repl_format_ts_human()'s own "%Y-%m-%dT%H:%M:%SZ" strftime
 * format (repl.c) -- e.g. "2026-09-02T03:29:38Z" -- never a specific
 * value, since the fixture's simulated wall-clock timestamp is not
 * deterministic across test runs (SCHED_SIM_START_TS_LATEST).
 */
static bool looks_like_iso8601_ts(const char* s)
{
    static const char pattern[] = "dddd-dd-ddTdd:dd:ddZ";
    size_t             i;

    if (s == NULL) {
        return false;
    }
    for (i = 0; pattern[i] != '\0'; i++) {
        if (s[i] == '\0') {
            return false;
        }
        if (pattern[i] == 'd') {
            if (!isdigit((unsigned char)s[i])) {
                return false;
            }
        } else if (s[i] != pattern[i]) {
            return false;
        }
    }
    return true;
}

/*
 * find_prompt_with_session - locates "(sim session=<expect_session> t=" in
 * `captured`, starting the search at `from` (pass `captured` itself to
 * search from the start), and confirms what follows is a well-formed
 * ISO8601 timestamp immediately closed by ") " -- i.e. the WHOLE prompt
 * shape design.md's repl_build_prompt() Service Interface specifies, not
 * just a coincidental substring match elsewhere in the output. Returns the
 * matching position (for ordering checks between two prompts) or NULL.
 */
static const char* find_prompt_with_session(const char*    captured,
                                            const char*    from,
                                            uint64_t       expect_session)
{
    char        needle[64];
    const char* pos;
    const char* ts_pos;

    snprintf(needle, sizeof(needle), "(sim session=%llu t=",
             (unsigned long long)expect_session);
    pos = strstr(from, needle);
    if (pos == NULL) {
        return NULL;
    }

    ts_pos = pos + strlen(needle);
    if (!looks_like_iso8601_ts(ts_pos)) {
        return NULL;
    }
    if (strncmp(ts_pos + 20, ") ", 2) != 0) {
        return NULL;
    }

    (void)captured;
    return pos;
}

/* ---------------------------------------------------------------------- */
/* Test 1: before any session has completed, the prompt is exactly the    */
/* plain "(sim) " -- unchanged from today. Expected to PASS already       */
/* (task 6.2's own regression-safety baseline, not new RED evidence).     */
/* ---------------------------------------------------------------------- */

static void test_prompt_is_plain_before_any_session_begins(void)
{
    char  replay_path[PATH_MAX];
    FILE* f;
    char* captured;
    char* banner_pos;
    char* prompt_pos;

    f = write_header_and_snapshot(replay_path, sizeof(replay_path),
                                  "_plain");
    TEST_ASSERT_NOT_NULL(f);

    append_queue_update_event(f, 1000ULL, "normal");
    append_jobs_push_event(f, 1500ULL, 6161ULL, "normal");
    append_replay_session(f, 2000ULL);
    append_end_of_stream_marker(f);
    fclose(f);

    /* `quit` immediately, before any `step`/`continue` is ever issued --
     * the pre-replay pause's prompt (the ONLY prompt shown by this script)
     * is captured while rs->session_count == 0. */
    captured = run_repl_via_pty(replay_path, "quit\n");

    banner_pos = strstr(captured, "[sim] ready. type 'help' for commands.");
    TEST_ASSERT_NOT_NULL_MESSAGE(banner_pos,
                                 "the pre-replay pause banner must still "
                                 "print (Requirement 5.5 changes only the "
                                 "prompt, not this banner)");

    prompt_pos = strstr(banner_pos, "(sim) ");
    TEST_ASSERT_NOT_NULL_MESSAGE(
        prompt_pos, "the first prompt, shown before any session has "
                    "begun, must be exactly the plain \"(sim) \" prompt");

    TEST_ASSERT_NULL_MESSAGE(
        strstr(captured, "(sim session="),
        "no session-aware prompt form may appear before any session has "
        "completed -- rs->session_count is still 0 for the whole script");

    free(captured);

    printf("test_prompt_is_plain_before_any_session_begins: PASS\n");
}

/* ---------------------------------------------------------------------- */
/* Test 2: once a session has completed, the prompt shows the correct     */
/* bare session count plus a well-formed ISO8601 timestamp -- proved in   */
/* BOTH step mode (the post-`step N` pause) and continue mode (the        */
/* post-replay-completion re-entry), in that order across one script, and */
/* WITHOUT altering `info summary`'s own existing output text (Cover      */
/* bullets 2, 3, 4). This is the test expected to FAIL (RED) against      */
/* today's static XCE_REPL_PROMPT -- the current prompt never contains    */
/* "session=" at all, so find_prompt_with_session() finds nothing for     */
/* either session count below.                                            */
/* ---------------------------------------------------------------------- */

static void test_prompt_shows_session_and_timestamp_in_step_and_continue_modes(
    void)
{
    char        replay_path[PATH_MAX];
    FILE*       f;
    char*       captured;
    const char* session1_pos;
    const char* session2_pos;

    f = write_header_and_snapshot(replay_path, sizeof(replay_path),
                                  "_sessioncount");
    TEST_ASSERT_NOT_NULL(f);

    append_queue_update_event(f, 1000ULL, "normal");
    append_jobs_push_event(f, 1500ULL, 7171ULL, "normal");
    append_replay_session(f, 2000ULL); /* session boundary #1 */
    append_jobs_push_event(f, 2500ULL, 7272ULL, "normal");
    append_replay_session(f, 3000ULL); /* session boundary #2 */
    append_end_of_stream_marker(f);
    fclose(f);

    /* "step 1" completes exactly session 1, pausing at a prompt that must
     * show session=1. "info summary" is issued while still at session=1,
     * to prove its own output text is unaffected by the prompt change
     * (Cover bullet 4). "continue" then runs session 2 to EOF, re-entering
     * the command loop via xce_repl_on_replay_complete() at a prompt that
     * must show session=2 (Cover bullet 3: continue mode). */
    captured = run_repl_via_pty(replay_path,
                                "step 1\ninfo summary\ncontinue\nquit\n");

    /* Cover bullet 4: "info summary"'s own existing output text, byte-
     * identical to what every other REPL test in this file family already
     * asserts for a one-job, still-pending fixture (e.g.
     * test_repl_execution_control.c's own "1 jobs (1 pending, 0 running)"
     * assertion) -- proves the prompt change does not alter a command's
     * own printed output, only the prompt line itself. */
    TEST_ASSERT_NOT_NULL_MESSAGE(
        strstr(captured, "1 jobs (1 pending, 0 running)"),
        "info summary's own output text must be byte-identical to today's "
        "-- the prompt change must not alter any command's own output");

    /* Cover bullet 2 + 3 (step mode): the prompt shown right after `step
     * 1` completes must include the bare session count (1, not 2) and a
     * well-formed ISO8601 timestamp. */
    session1_pos = find_prompt_with_session(captured, captured, 1);
    TEST_ASSERT_NOT_NULL_MESSAGE(
        session1_pos,
        "after `step 1` completes session 1, the prompt must read "
        "\"(sim session=1 t=<ISO8601-ts>) \" -- FAILS against today's "
        "static \"(sim) \" prompt (RED phase)");

    /* Cover bullet 3 (continue mode): the prompt shown after `continue`
     * drains session 2 to EOF and xce_repl_on_replay_complete() re-enters
     * the command loop must include session=2. */
    session2_pos = find_prompt_with_session(captured, session1_pos, 2);
    TEST_ASSERT_NOT_NULL_MESSAGE(
        session2_pos,
        "after `continue` drains session 2 to EOF, the post-replay-"
        "complete prompt must read \"(sim session=2 t=<ISO8601-ts>) \" -- "
        "FAILS against today's static \"(sim) \" prompt (RED phase)");

    TEST_ASSERT_TRUE_MESSAGE(
        session1_pos < session2_pos,
        "the session=1 prompt (after `step 1`) must appear strictly "
        "before the session=2 prompt (after `continue`) -- proves the "
        "count tracks progress, not a fixed/coincidental value");

    free(captured);

    printf("test_prompt_shows_session_and_timestamp_in_step_and_continue_"
          "modes: PASS\n");
}

int main(void)
{
    UNITY_BEGIN();

    RUN_TEST(test_prompt_is_plain_before_any_session_begins);
    RUN_TEST(test_prompt_shows_session_and_timestamp_in_step_and_continue_modes);

    return UNITY_END();
}