jpmschweitzerandClaude Opus 5
3b211a5450
The non-negotiable from D-263, pointed at its worst hiding place: a foreground command that swallows a failure at least does it in front of someone, while a background runner that reports "started" and loses the failure does it where nothing is watching. Testing the two timing cases the ticket names — fails before the parent exits, fails long after — needs a command slow enough to tell them apart, and every verb in reach finishes in milliseconds. So `reach dev selftest` exists: emits progress for N seconds, then optionally fails with a chosen code. A genuine diagnostic rather than a test hook, in the dev domain the map already planned, and the only way to answer "does streaming work here, can I tail it, does a failure survive detach" by observation instead of argument. The slow case is the one that proves the design. --detach returned in 75ms while the child ran six seconds, so the parent was demonstrably gone long before the child failed — and wait still relayed exit 7. That is the half of the recording path only this case reaches, and why T-1277 moved completion recording into the child. Also pinned: --detach exits 0 for starting and SAYS "not succeeded" in words, which the test asserts on rather than trusting the code to be read correctly; a failed job nobody waited on shows as failed in jobs list; and every event a detached job emits carries its job id. Closed T-1278's open gap in passing — jobs log --follow had never run against a genuinely long job because none existed. It now has: attached mid-flight, streamed the remaining steps live, and caught the final verdict after the job ended. Proven to fail by making effective_exit_code always return 0 — the trap itself. Both timing cases failed by name. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
21 lines
719 B
Python
21 lines
719 B
Python
"""Logic for the `dev` domain. Transport-agnostic (D-263)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from collections.abc import Iterator
|
|
|
|
|
|
def slow_work(seconds: float, steps: int = 10) -> Iterator[tuple[int, int]]:
|
|
"""Yield (step, total) over roughly `seconds`, so a caller can report progress.
|
|
|
|
A generator rather than something that takes a callback, because the service
|
|
must not know how progress is reported — that is the router's business, and
|
|
handing the service a printer is how a service learns it lives in a CLI.
|
|
"""
|
|
steps = max(1, steps)
|
|
interval = max(0.0, seconds) / steps
|
|
for step in range(1, steps + 1):
|
|
time.sleep(interval)
|
|
yield step, steps
|