-- End-to-end harness: the real `panto` binary, the real rock, a scripted
-- provider.
--
-- Every scenario builds a throwaway config layer under a fresh temp root, drops
-- e2e/fixture.lua and a symlink to this repo into an `extensions.paths`
-- directory, and runs `panto -p <script>` in a throwaway working directory. The
-- fixture protocol turns the prompt into a scripted turn (see e2e/fixture.lua),
-- so the primary emits exactly the tool batch a scenario needs and its children
-- answer deterministically. Nothing below the extension API is stubbed: real
-- agents, real luv parking, real JSONL on disk.
--
-- Why print mode rather than a PTY: `panto -p` builds the identical
-- SessionWorld the TUI does (agent, Lua interpreter, ExtHost, extension
-- discovery and activation, protocol router) and runs the turn on a pump thread
-- while the main thread owns the libuv loop — which is precisely the seam the
-- children's wake pipes and tool batches ride on. The only things it does not
-- provide are the TUI-only surfaces: `turn_start`/`turn_end`/`turn_interrupt`
-- never fire (no components, no Escape), so the progress cards and the
-- interrupt path degrade to no-ops here and stay covered by the rock specs.
--
-- Run: `mise run e2e` (or `lua e2e/run.lua [--only <name>] [--keep]`).

local function sh(text)
    return "'" .. tostring(text):gsub("'", "'\\''") .. "'"
end

local function popen_line(command)
    local pipe = io.popen(command, "r")
    if pipe == nil then
        return nil
    end
    local text = pipe:read("a")
    pipe:close()
    return (text or ""):gsub("%s+$", "")
end

-- Absolute: every scenario runs with its cwd inside a temp directory, so a
-- path relative to the invocation would stop resolving the moment a turn runs.
local REPO = popen_line("cd \"$(dirname " .. sh(arg[0]) .. ")/..\" && pwd")
assert(REPO ~= nil and REPO ~= "", "could not resolve the repository root")

local PANTO = os.getenv("PANTO_BIN")
if PANTO == nil or PANTO == "" then
    PANTO = REPO .. "/../pantograph/zig-out/bin/panto"
end

-- The luarocks tree panto bootstraps for itself, in the developer's real data
-- home. Every scenario gets a throwaway data home (so the auto-generated base
-- config/models layer cannot drift a scenario's assertions from one machine to
-- the next) with this tree symlinked in, which is the one thing there too
-- expensive to rebuild per run.
local REAL_ROCKS = (os.getenv("XDG_DATA_HOME") or (os.getenv("HOME") or "") .. "/.local/share")
    .. "/panto/rocks"

-- A wedged child is the failure this harness exists to catch; without a cap it
-- would present as a hung suite instead of a diff.
local TIMEOUT = popen_line("command -v timeout || command -v gtimeout || true")
local TIMEOUT_SECONDS = tonumber(os.getenv("PANTO_E2E_TIMEOUT") or "") or 120

local only = nil
local keep = false
for index = 1, #arg do
    if arg[index] == "--only" then
        only = arg[index + 1]
    elseif arg[index] == "--keep" then
        keep = true
    end
end

-- ---------------------------------------------------------------------------
-- Shell and filesystem helpers
-- ---------------------------------------------------------------------------

local function run_shell(command)
    local ok, how, code = os.execute(command)
    if ok == true then
        return 0
    end
    if how == "exit" then
        return code or 1
    end
    return 128 + (code or 0)
end

local function mkdirp(path)
    assert(run_shell("mkdir -p " .. sh(path)) == 0, "mkdir failed: " .. path)
end

local function write_file(path, text)
    local fh = assert(io.open(path, "w"))
    fh:write(text)
    fh:close()
end

local function read_file(path)
    local fh = io.open(path, "r")
    if fh == nil then
        return nil
    end
    local text = fh:read("a")
    fh:close()
    return text
end

local function list_dir(path)
    local out = popen_line("ls -1 " .. sh(path) .. " 2>/dev/null")
    local names = {}
    for name in (out or ""):gmatch("[^\n]+") do
        names[#names + 1] = name
    end
    table.sort(names)
    return names
end

local function lines_of(text)
    local out = {}
    for line in (text or ""):gmatch("[^\n]+") do
        out[#out + 1] = line
    end
    return out
end

-- ---------------------------------------------------------------------------
-- Assertions
-- ---------------------------------------------------------------------------

local failures = {}
local checks = 0
local scenario_name = "?"
local scenario_context = {}

local function fail(what, detail)
    failures[#failures + 1] = { scenario = scenario_name, what = what, detail = detail }
    io.write(string.format("  FAIL  %s\n", what))
    if detail ~= nil and detail ~= "" then
        for _, line in ipairs(lines_of(detail)) do
            io.write("        " .. line .. "\n")
        end
    end
end

local function pass(what)
    checks = checks + 1
    io.write(string.format("  ok    %s\n", what))
end

-- Truncated context for a failing assertion: the haystack is often a whole
-- turn's stdout or a JSONL file, and dumping all of it buries the diff.
local function excerpt(text, limit)
    text = tostring(text)
    limit = limit or 1600
    if #text <= limit then
        return text
    end
    return text:sub(1, limit) .. "\n... (" .. (#text - limit) .. " more bytes)"
end

local function check(what, ok, detail)
    if ok then
        pass(what)
    else
        fail(what, detail)
    end
    return ok
end

local function check_contains(what, haystack, needle)
    return check(what, (haystack or ""):find(needle, 1, true) ~= nil,
        "expected to find:\n" .. needle .. "\nin:\n" .. excerpt(haystack))
end

local function check_absent(what, haystack, needle)
    return check(what, (haystack or ""):find(needle, 1, true) == nil,
        "expected NOT to find:\n" .. needle .. "\nin:\n" .. excerpt(haystack))
end

local function check_equal(what, got, want)
    return check(what, got == want,
        "got:  " .. tostring(got) .. "\nwant: " .. tostring(want))
end

local function count_occurrences(haystack, needle)
    local seen, start = 0, 1
    while true do
        local from, to = (haystack or ""):find(needle, start, true)
        if from == nil then
            return seen
        end
        seen = seen + 1
        start = to + 1
    end
end

-- ---------------------------------------------------------------------------
-- The throwaway panto installation
-- ---------------------------------------------------------------------------

local tmp_root = popen_line("mktemp -d " .. sh((os.getenv("TMPDIR") or "/tmp"):gsub("/$", "") .. "/panto-subagents-e2e-XXXXXX"))
assert(tmp_root ~= nil and tmp_root ~= "", "could not create a temp root")

local CONFIG = [[
[api]
timeout = 60
retries = 0

# The children park the primary's tool batch for as long as they take; the
# watchdog would otherwise fail a batch this harness deliberately holds open.
[tools]
batch_warn_seconds = 0
batch_kill_seconds = 0

[providers.fixture-provider]
protocol = "fixture"

[defaults]
model = "fixture-provider:mini"
reasoning = "tiny"

# The second path is this repository, added exactly the way a local checkout is
# meant to be: its init.lua is a single-file source whose package root is the
# checkout, so `require("subagents.jobs")` resolves to subagents/jobs.lua just
# as it does from an installed rock.
#
# A pure whitelist: the real base layer's shipped extensions are still
# discovered and evaluated, but only these two activate, so a child's inherited
# tool set is exactly what this fixture registers.
[extensions]
paths = ["%s", "%s"]
deny = ["**"]
allow = ["subagents", "fixture"]
]]

local MODELS = [[
[fixture-provider.mini]
model = "mini-wire"

[fixture-provider.beta]
model = "beta-wire"
]]

local PROFILE_ALPHA = [[
---
name: alpha
description: the alpha profile
---

ALPHA-PROFILE-BODY
]]

local PROFILE_BETA = [[
---
name: beta
description: the beta profile
model: fixture-provider:beta
reasoning: deep
---

BETA-PROFILE-BODY
]]

local WORKFLOW_DIAMOND = [==[
name = "diamond"
description = "one root, two independent branches, one sink"

[[steps]]
id = "root"
agent = "alpha"
prompt = "say ROOT-OUT"

[[steps]]
id = "left"
agent = "alpha"
prompt = "tag LEFT ROOT-OUT"
needs = ["root"]

[[steps]]
id = "right"
agent = "beta"
prompt = "tag RIGHT ROOT-OUT"
needs = ["root"]

[[steps]]
id = "sink"
agent = "alpha"
prompt = "body"
needs = ["left", "right"]
]==]

-- One installation per scenario: its own config layer, working directory,
-- sessions base and trace file, so a scenario can assert on "every session in
-- this directory" without seeing another's.
local function make_install(name)
    local root = tmp_root .. "/" .. name
    local ext = root .. "/ext"
    local work = root .. "/work"
    mkdirp(root .. "/config/panto")
    mkdirp(root .. "/data/panto")
    mkdirp(ext)
    mkdirp(work .. "/.panto/agents")
    mkdirp(work .. "/.panto/workflows")
    mkdirp(root .. "/sessions")
    assert(run_shell("ln -sfn " .. sh(REAL_ROCKS) .. " " .. sh(root .. "/data/panto/rocks")) == 0)

    write_file(root .. "/config/panto/config.toml", CONFIG:format(ext, REPO))
    write_file(root .. "/config/panto/models.toml", MODELS)
    write_file(work .. "/.panto/agents/alpha.md", PROFILE_ALPHA)
    write_file(work .. "/.panto/agents/beta.md", PROFILE_BETA)
    write_file(work .. "/.panto/workflows/diamond.toml", WORKFLOW_DIAMOND)
    -- The project layer replaces the built-in system prompt outright, so the
    -- context a child inherits is exactly this marker.
    write_file(work .. "/.panto/SYSTEM.md", "PRIMARY-CONTEXT-MARK\n")

    assert(run_shell("cp " .. sh(REPO .. "/e2e/fixture.lua") .. " " .. sh(ext .. "/fixture.lua")) == 0)

    return {
        name = name,
        root = root,
        work = work,
        sessions = root .. "/sessions",
        trace = root .. "/trace.txt",
        config_home = root .. "/config",
        data_home = root .. "/data",
        runs = 0,
    }
end

-- panto's per-cwd session directory below the PANTO_SESSION_DIR base.
local function session_dir(install)
    local entries = list_dir(install.sessions)
    if #entries ~= 1 then
        return nil, string.format("expected one per-cwd session directory under %s, saw %d",
            install.sessions, #entries)
    end
    return install.sessions .. "/" .. entries[1]
end

-- The binary, in this installation's environment. Every layer panto reads is
-- redirected into the temp root — base included, so the developer's own
-- providers, model aliases and extension paths cannot reach a scenario; only
-- the bootstrapped luarocks tree is borrowed, by symlink.
local function invocation(install)
    local parts = {
        "cd", sh(install.work), "&&",
        "XDG_CONFIG_HOME=" .. sh(install.config_home),
        "XDG_DATA_HOME=" .. sh(install.data_home),
        "PANTO_SESSION_DIR=" .. sh(install.sessions),
        "PANTO_E2E_TRACE=" .. sh(install.trace),
        "PANTO_E2E_ROCKS=" .. sh(REPO .. "/.rocks"),
    }
    if TIMEOUT ~= nil and TIMEOUT ~= "" then
        parts[#parts + 1] = sh(TIMEOUT)
        parts[#parts + 1] = tostring(TIMEOUT_SECONDS)
    end
    parts[#parts + 1] = sh(PANTO)
    return parts
end

-- Run one turn. `resume` is a primary session id (or nil for a new session).
local function turn(install, script, resume)
    install.runs = install.runs + 1
    local log = string.format("%s/run-%d.stderr", install.root, install.runs)
    local parts = invocation(install)
    if resume ~= nil then
        parts[#parts + 1] = "--resume"
        parts[#parts + 1] = sh(resume)
    end
    parts[#parts + 1] = "-p"
    parts[#parts + 1] = sh(script)
    parts[#parts + 1] = "2>" .. sh(log)

    local command = table.concat(parts, " ")
    local pipe = assert(io.popen(command, "r"))
    local out = pipe:read("a") or ""
    local ok, how, code = pipe:close()

    local status = (ok == true) and 0 or (how == "exit" and (code or 1) or 128 + (code or 0))
    scenario_context = {
        command = command,
        stdout = out,
        stderr = read_file(log) or "",
        status = status,
    }
    -- A turn that did not finish makes every assertion after it noise, so this
    -- aborts the scenario with the one failure that matters.
    if status == 124 then
        error(string.format("panto timed out after %ds\nstdout so far:\n%s",
            TIMEOUT_SECONDS, excerpt(out)), 0)
    elseif status ~= 0 then
        error(string.format("panto exited %d\nstdout:\n%s", status, excerpt(out)), 0)
    end
    io.write(string.format("  ..    turn %d (%s): %d bytes of output\n",
        install.runs, resume and ("resume " .. resume:sub(1, 8)) or "new", #out))
    if os.getenv("PANTO_E2E_VERBOSE") then
        for _, line in ipairs(lines_of(out)) do
            io.write("        | " .. line .. "\n")
        end
    end
    return out, status
end

-- ---------------------------------------------------------------------------
-- Reading the store back
-- ---------------------------------------------------------------------------

-- Per-message metadata is stored as a JSON *string* inside the JSONL record, so
-- its quotes are escaped one level deeper than the record's own fields. These
-- assertions read the bytes on disk, not a decoded view, so they escape too.
local function meta(needle)
    return (needle:gsub('"', '\\"'))
end

local function json_string(text)
    return '"' .. tostring(text)
        :gsub('\\', '\\\\')
        :gsub('"', '\\"')
        :gsub('\r', '\\r')
        :gsub('\n', '\\n') .. '"'
end

local function jsonl_ids(dir)
    local ids = {}
    for _, name in ipairs(list_dir(dir)) do
        local stem = name:match("^(.+)%.jsonl$")
        if stem then
            ids[#ids + 1] = stem
        end
    end
    return ids
end

-- What `panto sessions` shows for this working directory: the user-facing view
-- that must never mention a child.
local function sessions_listing(install)
    local parts = invocation(install)
    parts[#parts + 1] = "sessions"
    parts[#parts + 1] = "2>&1"
    return popen_line(table.concat(parts, " ")) or ""
end

local function child_dir(install, primary_id)
    local dir, err = session_dir(install)
    if dir == nil then
        return nil, err
    end
    return dir .. "/subagents/" .. primary_id
end

-- Every `subagents.run` result block, keyed in stdout order.
local function result_blocks(out)
    local blocks = {}
    for block in ("\n" .. out):gmatch("\nid: [^\n]*\nagent: [^\n]*\nstatus: [^\n]*\nresumable: [^\n]*\n%-%-%- output %-%-%-\n") do
        blocks[#blocks + 1] = block
    end
    return blocks
end

-- ---------------------------------------------------------------------------
-- Scenarios
-- ---------------------------------------------------------------------------

local scenarios = {}

local function scenario(name, body)
    scenarios[#scenarios + 1] = { name = name, body = body }
end

-- 1. One tool batch, seven calls: two children that must overlap, three that
--    report what they were handed, one that fails mid-stream, and one unknown
--    profile that never becomes a child. Six jobs against a bound of five also
--    means one of them starts only when a slot frees.
scenario("parallel-batch", function()
    local install = make_install("parallel-batch")
    local script = table.concat({
        [[tool subagents.run {"agent":"alpha","prompt":"tool pair {\"text\":\"A\"}"}]],
        [[tool subagents.run {"agent":"beta","prompt":"tool pair {\"text\":\"B\"}"}]],
        [[tool subagents.run {"agent":"alpha","prompt":"tool whoami {}"}]],
        [[tool subagents.run {"agent":"beta","prompt":"tools"}]],
        [[tool subagents.run {"agent":"alpha","prompt":"sys PRIMARY-CONTEXT-MARK"}]],
        [[tool subagents.run {"agent":"alpha","prompt":"err BOOM-CHILD"}]],
        [[tool subagents.run {"agent":"ghost","prompt":"say never"}]],
    }, " && ")
    local out = turn(install, script)

    check_equal("seven results came back",
        #result_blocks(out) + count_occurrences(out, "Error: unknown agent"), 7)
    check_contains("the unknown profile failed by name", out,
        "Error: unknown agent 'ghost'; known: alpha, beta")
    check_contains("child A reported its rendezvous", out, "pair:A")
    check_contains("child B reported its rendezvous", out, "pair:B")
    check_contains("the primary's system context did not reach a child", out, "sys-missing PRIMARY-CONTEXT-MARK")

    -- One child's stream failed; its five siblings settled on their own terms.
    check_equal("five children completed", count_occurrences(out, "status: completed"), 5)
    check_equal("exactly one child failed", count_occurrences(out, "status: failed"), 1)
    check_contains("the failure carries the provider's message", out, "BOOM-CHILD")
    -- It died before its first assistant message, so the store flushed nothing
    -- and the result must not promise a continuation.
    check_contains("a child with no durable file is not resumable", out,
        "status: failed\nresumable: false")

    -- Lane identity: `panto.ext.agent` inside a child's tool handler is that
    -- child, so the id it printed must be the id the result block reports.
    local block_id, whoami_id = out:match("id: ([%x%-]+)\nagent: alpha\nstatus: completed\nresumable: true\n%-%-%- output %-%-%-\n([%x%-]+)")
    check("a child's whoami is its own session id",
        block_id ~= nil and block_id == whoami_id,
        "block id: " .. tostring(block_id) .. "\nwhoami:   " .. tostring(whoami_id))

    -- Tool inheritance: everything the primary has except the delegation tools.
    local offered = out:match("\n%-%-%- output %-%-%-\n(echo[^\n]*)")
    check_equal("a child is offered exactly the primary's non-delegation tools", offered, "echo,pair,whoami")

    -- Overlap: both children entered `pair` before either left it. A serialized
    -- pair would read enter/leave/enter/leave.
    --
    -- This is also the proof that jobs.lua really parked. `pair` runs on the
    -- owner thread, so the children's tool batches are only served while the
    -- primary's own handler is suspended and the loop is running. Had the wake
    -- pipe or its uv poll failed to arm, `jobs.await` would have fallen back to
    -- its drain-and-sleep loop, never returned to the loop, and these two
    -- handlers would never have run at all.
    local trace = lines_of(read_file(install.trace) or "")
    local order = {}
    for _, line in ipairs(trace) do
        if line:match("^enter ") or line:match("^leave ") then
            order[#order + 1] = line
        end
    end
    check("two children were in flight at once",
        #order == 4 and order[1]:match("^enter") and order[2]:match("^enter")
            and order[3]:match("^leave") and order[4]:match("^leave"),
        "rendezvous trace:\n" .. table.concat(order, "\n"))

    -- Bytes on disk.
    local dir = assert(session_dir(install))
    local top = jsonl_ids(dir)
    check_equal("the primary store holds exactly one top-level session", #top, 1)
    local primary_id = top[1]

    local children = assert(child_dir(install, primary_id))
    local child_ids = jsonl_ids(children)
    -- Six children started; the one whose stream failed left no file behind,
    -- which is the same fact `resumable: false` reported above.
    check_equal("the five children that answered persisted under subagents/<primary>",
        #child_ids, 5)
    local top_text = table.concat(top, "\n")
    local top_level = {}
    for _, id in ipairs(child_ids) do
        if top_text:find(id, 1, true) then
            top_level[#top_level + 1] = id
        end
    end
    check("no child id appears as a top-level session", #top_level == 0,
        "found at the top level:\n" .. table.concat(top_level, "\n"))

    -- The picker's own view, not just the directory: `list`/`resolve` look at
    -- direct .jsonl children only, so a child is invisible to the user.
    -- Row-counting rather than id-matching: `sessions` prints short ids, and
    -- every UUIDv7 minted in the same millisecond shares its first eight
    -- characters, so a substring test would pass for the wrong reason.
    local listing = sessions_listing(install)
    local rows = {}
    for _, line in ipairs(lines_of(listing)) do
        if line:match("^%x%x%x%x%x%x%x%x%s") then
            rows[#rows + 1] = line
        end
    end
    check_equal("panto sessions lists one session, not seven", #rows, 1)
    check("panto sessions lists the primary",
        rows[1] ~= nil and rows[1]:sub(1, 8) == primary_id:sub(1, 8),
        "listing:\n" .. excerpt(listing))

    -- The manifest rides on the profile system message; the per-turn model and
    -- reasoning ride on the user message.
    local seen_manifest, seen_turn, seen_beta = 0, 0, 0
    for index, id in ipairs(child_ids) do
        local who = "child " .. index
        local text = read_file(children .. "/" .. id .. ".jsonl") or ""
        for _, line in ipairs(lines_of(text)) do
            if line:find(meta('"subagents"'), 1, true) and line:find(meta('"owner":"' .. primary_id .. '"'), 1, true) then
                seen_manifest = seen_manifest + 1
                check(who .. ": the manifest is on a system message",
                    line:find('"role":"system"', 1, true) ~= nil, excerpt(line, 400))
                check(who .. ": the manifest names the profile",
                    line:find(meta('"agent":"alpha"'), 1, true) ~= nil
                        or line:find(meta('"agent":"beta"'), 1, true) ~= nil, excerpt(line, 400))
            elseif line:find(meta('"subagents"'), 1, true) and line:find(meta('"model":'), 1, true) then
                seen_turn = seen_turn + 1
                check(who .. ": per-turn metadata is on a user message",
                    line:find('"role":"user"', 1, true) ~= nil, excerpt(line, 400))
                if line:find(meta('"model":"fixture-provider:beta"'), 1, true) then
                    seen_beta = seen_beta + 1
                    check(who .. ": the beta profile's reasoning came with its model",
                        line:find(meta('"reasoning":"deep"'), 1, true) ~= nil, excerpt(line, 400))
                else
                    check(who .. ": an alpha child inherited the primary's model and reasoning",
                        line:find(meta('"model":"fixture-provider:mini"'), 1, true) ~= nil
                            and line:find(meta('"reasoning":"tiny"'), 1, true) ~= nil,
                        excerpt(line, 400))
                end
            end
        end
        check_contains(who .. ": the child-role instruction was seeded", text,
            "You are a subagent working inside another agent's session")
        local primary_context_in_system = false
        for _, line in ipairs(lines_of(text)) do
            if line:find('"role":"system"', 1, true)
                and line:find("PRIMARY-CONTEXT-MARK", 1, true) then
                primary_context_in_system = true
                break
            end
        end
        check(who .. ": the primary's system context was not seeded",
            not primary_context_in_system, excerpt(text, 400))
    end
    check_equal("every child carries one manifest", seen_manifest, 5)
    check_equal("every child recorded one turn's model/reasoning", seen_turn, 5)
    check_equal("the two beta children ran on the profile's model", seen_beta, 2)
end)

-- 2. A one-off child can define its own system prompt without a profile or a
--    workflow wrapper.
scenario("inline-run", function()
    local install = make_install("inline-run")
    local out = turn(install,
        [[tool subagents.run {"system_prompt":"INLINE-RUN-MARK","prompt":"sys INLINE-RUN-MARK"}]])
    check_contains("the inline child completed", out, "status: completed")
    check_contains("the inline prompt reached the child", out, "sys-found INLINE-RUN-MARK")
    check_contains("the inline child has a stable result label", out, "agent: subagent")

    local dir = assert(session_dir(install))
    local primary_id = jsonl_ids(dir)[1]
    local children = jsonl_ids(assert(child_dir(install, primary_id)))
    check_equal("the inline child persisted normally", #children, 1)
    local text = read_file(assert(child_dir(install, primary_id)) .. "/" .. children[1] .. ".jsonl") or ""
    check_contains("the inline manifest is replay-visible", text, meta('"inline":true'))
end)

-- 3. A child outlives its process: a second `panto --resume` continues it, and a
--    third, unrelated primary cannot.
scenario("resume", function()
    local install = make_install("resume")
    local first = turn(install,
        [[tool subagents.run {"agent":"alpha","model":"fixture-provider:beta","reasoning":"deep","prompt":"say HELLO-ONE"}]])
    check_contains("the first turn completed", first, "status: completed")
    check_contains("the first turn is resumable", first, "resumable: true")
    check_contains("the child answered", first, "HELLO-ONE")

    local child_id = first:match("id: ([%x%-]+)")
    check("the first turn reported a child id", child_id ~= nil, excerpt(first))
    if child_id == nil then
        return
    end

    local dir = assert(session_dir(install))
    local primary_id = jsonl_ids(dir)[1]
    local children = assert(child_dir(install, primary_id))

    local second = turn(install,
        string.format([[tool subagents.run {"id":"%s","prompt":"find HELLO-ONE"}]], child_id),
        primary_id)
    check_contains("the resumed child saw its own history", second, "found HELLO-ONE")
    check_contains("the resumed child completed", second, "status: completed")
    check_contains("the resumed child re-identified itself from the manifest", second, "agent: alpha")
    check_contains("the resumed child kept its id", second, "id: " .. child_id)

    local text = read_file(children .. "/" .. child_id .. ".jsonl") or ""
    check_contains("the resumed child records its owning outer tool call", text,
        meta('"tool_call_id"'))
    local turns = 0
    for _, line in ipairs(lines_of(text)) do
        if line:find(meta('"subagents"'), 1, true) and line:find(meta('"model":'), 1, true) then
            turns = turns + 1
            -- The second turn passed no overrides: these values came back out
            -- of the stored conversation, not out of the primary's defaults.
            check("turn " .. turns .. " recorded the call's model override",
                line:find(meta('"model":"fixture-provider:beta"'), 1, true) ~= nil
                    and line:find(meta('"reasoning":"deep"'), 1, true) ~= nil,
                excerpt(line, 400))
        end
    end
    check_equal("the child's file records two turns", turns, 2)
    check_equal("one conversation, one file", #jsonl_ids(children), 1)

    -- A different primary session shares neither the catalog directory nor the
    -- id space, so the same id is simply not there.
    local third = turn(install, string.format([[tool subagents.run {"id":"%s","prompt":"say nope"}]], child_id))
    check_contains("another primary cannot resolve the child", third,
        string.format("Error: unknown subagent id '%s' for this session", child_id))
end)

-- 4. A discovered TOML workflow: a diamond, so both branches run concurrently
--    off one root and the sink sees both labelled outputs.
scenario("workflow-diamond", function()
    local install = make_install("workflow-diamond")
    local out = turn(install, [[tool subagents.workflow {"name":"diamond","prompt":"DIAMOND-INPUT"}]])

    check_equal("only the sink is terminal", count_occurrences("\n" .. out, "\nstep: "), 1)
    check_contains("the sink is the terminal node", out, "step: sink")
    check_contains("the sink completed", out, "step: sink\nstatus: completed")
    check_contains("every step received the workflow input", out, "## Workflow input\n\nDIAMOND-INPUT")
    check_contains("the left branch saw the root's output", out, "## Output of left\n\nLEFT:found")
    check_contains("the right branch saw the root's output", out, "## Output of right\n\nRIGHT:found")

    local dir = assert(session_dir(install))
    local primary_id = jsonl_ids(dir)[1]
    check_equal("four steps became four children", #jsonl_ids(assert(child_dir(install, primary_id))), 4)
end)

-- 5. The restricted `subagents.lua` sandbox: a one-shot structured worker whose
--    decoded output drives the fan-out.
scenario("sandbox-fanout", function()
    local install = make_install("sandbox-fanout")
    local source = table.concat({
        "return subagents.workflow(function(ctx)",
        "  local split = ctx:agent({",
        "    name = 'split',",
        "    agent = 'alpha',",
        [[    prompt = 'emit emit_result {\"items\":[\"X\",\"Y\"]}',]],
        "    output = { description = 'the work items', schema = { type = 'object',",
        "      required = { 'items' },",
        "      properties = { items = { type = 'array', items = { type = 'string' } } } } },",
        "  }):await()",
        "  local handles = {}",
        "  for _, item in ipairs(split.output.items) do",
        "    handles[#handles + 1] = ctx:agent({ name = 'item-' .. item, agent = 'beta', prompt = 'say ITEM-' .. item })",
        "  end",
        "  local results = ctx:await(handles, 'all')",
        "  return results[1].output .. '\\\\n' .. results[2].output",
        "end)",
    }, "\\n")
    local out = turn(install, string.format(
        [[tool subagents.lua {"source":"%s"}]], source))

    check_contains("the workflow id returned immediately", out, "workflow-")

    -- The one-shot worker is ephemeral: only the two conversational children
    -- have a durable file.
    local dir = assert(session_dir(install))
    local primary_id = jsonl_ids(dir)[1]
    local children = assert(child_dir(install, primary_id))
    local ids = jsonl_ids(children)
    check_equal("the structured worker left no session behind", #ids, 2)
    local persisted = ""
    for _, id in ipairs(ids) do persisted = persisted .. (read_file(children .. "/" .. id .. ".jsonl") or "") end
    check_contains("the first fan-out branch ran", persisted, "ITEM-X")
    check_contains("the second fan-out branch ran", persisted, "ITEM-Y")
end)

-- 6. An inline profile in the dynamic workflow gets a durable marker and
--    body, so restart replay can show the requested system prompt without
--    exposing discovered profile prompts.
scenario("inline-replay-manifest", function()
    local install = make_install("inline-replay-manifest")
    local source = [[return subagents.workflow(function(ctx)
        local result = ctx:agent({ name = 'inline-work', agent = 'inline', prompt = 'say INLINE-CHILD' }):await()
        return result.output
    end)]]
    local script = string.format(
        'tool subagents.lua {"source":%s,"agents":[{"name":"inline","system_prompt":"INLINE-SYSTEM"}]}',
        json_string(source))
    local out = turn(install, script)
    check_contains("the inline workflow id returned", out, "workflow-")

    local dir = assert(session_dir(install))
    local primary_id = jsonl_ids(dir)[1]
    local children = assert(child_dir(install, primary_id))
    local ids = jsonl_ids(children)
    check_equal("one inline workflow child persisted", #ids, 1)
    local text = read_file(children .. "/" .. ids[1] .. ".jsonl") or ""
    check_contains("the inline manifest is marked", text, meta('"inline":true'))
    check_contains("the inline system prompt remains durable", text, "INLINE-SYSTEM")
end)

-- 7. The catalog tool, in its three shapes, against a provider whose reasoning
--    levels come from the protocol rather than models.toml.
scenario("catalog", function()
    local install = make_install("catalog")
    local out = turn(install, table.concat({
        [[tool subagents.models {}]],
        [[tool subagents.models {"model":"fixture-provider:beta"}]],
        [[tool subagents.models {"query":"beta"}]],
        [[tool subagents.models {"model":"fixture-provider:nope"}]],
    }, " && "))

    check_contains("the overview reports the inherited model", out, "inherited model: fixture-provider:mini")
    check_contains("the overview reports the inherited reasoning", out, "inherited reasoning: tiny")
    check_contains("the overview counts the fixture provider's models", out, "fixture-provider (")
    check_contains("the exact lookup resolves the wire name", out, "wire model: beta-wire")
    check_contains("the exact lookup reports the protocol's effort levels", out, "reasoning levels: tiny, deep")
    check_contains("the search found the model", out, "1 match(es):")
    check_contains("an unknown model is reported, not resolved", out,
        "No configured model matches 'fixture-provider:nope'")
end)

-- ---------------------------------------------------------------------------
-- Main
-- ---------------------------------------------------------------------------

io.write("panto:  " .. PANTO .. "\n")
io.write("root:   " .. tmp_root .. "\n")
if run_shell("test -x " .. sh(PANTO)) ~= 0 then
    io.write("\nerror: no panto binary at " .. PANTO ..
        "\n       build it: cd ../pantograph && mise exec -- zig build\n")
    os.exit(1)
end
-- Each scenario's data home is a fresh directory, so a missing tree here is not
-- a slow first run — it is every scenario bootstrapping luarocks from scratch.
if run_shell("test -d " .. sh(REAL_ROCKS)) ~= 0 then
    io.write("\nerror: no bootstrapped luarocks tree at " .. REAL_ROCKS ..
        "\n       create it once: " .. PANTO .. " bootstrap\n")
    os.exit(1)
end
-- Without lyaml the profiles parse as body-only, every model/reasoning
-- assertion drifts, and the diff blames the wrong thing. Say so instead.
if run_shell("test -d " .. sh(REPO .. "/.rocks/share/lua/5.4/lyaml")) ~= 0 then
    io.write("\nerror: no ./.rocks tree with lyaml (the rock reads profile frontmatter with it)" ..
        "\n       install it: mise run deps\n")
    os.exit(1)
end

local started_all = os.time()
for _, entry in ipairs(scenarios) do
    if only == nil or only == entry.name then
        scenario_name = entry.name
        io.write("\n== " .. entry.name .. "\n")
        local ok, err = pcall(entry.body)
        if not ok then
            fail("the scenario raised", tostring(err) .. "\nlast command:\n" ..
                tostring(scenario_context.command) .. "\nstderr:\n" ..
                excerpt(scenario_context.stderr))
        end
    end
end

io.write(string.format("\n%d checks, %d failures, %ds wall\n", checks, #failures, os.time() - started_all))
if #failures > 0 then
    io.write("\nfailed:\n")
    for _, entry in ipairs(failures) do
        io.write(string.format("  %s: %s\n", entry.scenario, entry.what))
    end
    io.write("\nartifacts kept at " .. tmp_root .. "\n")
    os.exit(1)
end

if keep then
    io.write("artifacts kept at " .. tmp_root .. "\n")
else
    run_shell("rm -rf " .. sh(tmp_root))
end
os.exit(0)