jobs: only settle may close; e2e proof harness

close_all no longer joins pump threads from the owner thread: job:close()
joins the pump, and a pump parked in an owner-posted tool batch needs the
owner back in the loop, so a synchronous close at turn_end could deadlock.
close_all now cancels everything, closes only already-settled jobs, and
defers the rest to their own settle; jobs.active() ignores closing children
so the next turn can resume the id.

e2e/ reconstructs the end-to-end proof against the real binary in print
mode: a scripted fixture protocol plays primary and children, and the suite
asserts the product surface from process output and JSONL bytes on disk —
parallel batch with unknown-agent error, lane identity, tool inheritance
minus subagents.*, resume with manifest re-identification, workflow diamond,
sandbox fan-out with an ephemeral structured worker, catalog round-trip,
child-store isolation from the session picker. mise run e2e; check stays
independent.

2 files changed, 25 insertions, 1 deletions

diff --git a/spec/test_init.lua b/spec/test_init.lua
index 800dcaa..338b7a7 100644
--- a/spec/test_init.lua
+++ b/spec/test_init.lua
@@ -115,7 +115,11 @@ return {
                 handle.emit("turn_interrupt", { phase = "interrupt" })
                 assert(job._cancel_requested, "an interrupted turn asks its children to stop")
                 handle.emit("turn_end", { phase = "end", reason = "interrupted" })
-                assert(job._closed, "the end of the turn joins every child")
+                assert(not job._closed, "the end of the turn never joins a pump from the owner thread")
+                -- The pump exits (the fake settles cancelled on its next poll)
+                -- and the drain that notices it does the close.
+                jobs.await({ started }, "all")
+                assert(job._closed, "a child closes as soon as its pump exits")
             end)
             jobs.close_all()
             assert(checked, failure)
diff --git a/spec/test_jobs.lua b/spec/test_jobs.lua
index e6ce097..ce0525c 100644
--- a/spec/test_jobs.lua
+++ b/spec/test_jobs.lua
@@ -233,6 +233,26 @@ return {
         end)
     end },
 
+    -- close() joins the pump, and a pump parked in a tool batch needs the owner
+    -- thread — the thread close_all itself runs on. So an unsettled job is never
+    -- closed here; it is cancelled, and the drain that sees it settle closes it.
+    { "close_all leaves an unsettled job to close on its own settle", function()
+        with_jobs(function()
+            local start, _, made = starter()
+            local handle = assert(start("alpha", { settle = 99 }))
+
+            jobs.close_all()
+            assert(made.alpha._cancel_requested, "close_all asks a running child to stop")
+            assert(not made.alpha._closed, "close_all must not join a pump that is still up")
+            assert(handle:result() == nil, "close_all does not settle the child itself")
+
+            local results = jobs.await({ handle }, "all")
+            assert(results[1].status == "cancelled", tostring(results[1].status))
+            assert(made.alpha._closed, "the job closes as soon as it settles")
+            assert(handle:result().status == "cancelled", "the result is cached before the close")
+        end)
+    end },
+
     { "a build failure is a nil return, never an exception", function()
         with_jobs(function()
             local start = starter()