HomerandClaude Opus 5
92070d7879
None of these fails loudly and two of them correct themselves if you reload, which is why five were found by reading rather than by anybody reporting them. The reply that lost its author is the one worth knowing about: the frame that replaces a bubble when a reply lands was looking the models up as nobody, and "no user" answers "no models" rather than "all models" -- so every finished reply swapped the model's avatar for the plain mark and put the instance's name where the model's should be, until the next page load put it back. Beside it: a concurrency quota enforced on two of the six paths that start a reply, including neither of the two most used; a custom theme whose success and warning colours moved the text and left the background behind; a phone shell sized to one viewport inside a document sized to another, which is the reported scroll past the bottom of Settings; a whole conversation's Markdown rendered on every page load and read by nothing; a skip guard inert since it was written; and an endpoint nothing has ever called. The scroll fix folded five near-identical scroller rules into one, which is also where the containment they were all missing now lives. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
611 lines
22 KiB
Python
611 lines
22 KiB
Python
"""Background jobs, exercised against a real local shell.
|
|
|
|
The sshd fixture elsewhere is a *fake* shell (it echoes the command), which
|
|
cannot run `setsid`, `base64`, a wait loop, or an exit-file. So these run the
|
|
wrappers through the machine's own `/bin/sh` with the same contract
|
|
`SshExecutor.run` has — cd-prefix, interleaved output, a timeout that leaves the
|
|
detached child alive — because the shell logic is the whole of the risk here.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import shutil
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from lembas.services import tools as tools_service
|
|
from lembas.services.agent import jobs, policy
|
|
from lembas.services.agent import ssh as ssh_service
|
|
from lembas.services.agent.base import ExecRequest, ExecResult, clean_output
|
|
from lembas.services.agent.session import AgentContext
|
|
from lembas.services.agent.tools import _run_shell
|
|
|
|
# A list, because there are two of them and `pytestmark = ...` twice is not two
|
|
# marks -- the second binding replaces the first, silently. It did, for the
|
|
# whole life of this file: the guard below was written, read as present, and
|
|
# never once applied, so a host without `setsid` got a module that errored
|
|
# instead of the skip somebody had taken the trouble to write.
|
|
pytestmark = [
|
|
pytest.mark.skipif(
|
|
shutil.which("setsid") is None or shutil.which("base64") is None,
|
|
reason="needs setsid and base64 (Linux)",
|
|
),
|
|
# Stands up something real -- see the `slow` marker in pyproject.toml.
|
|
pytest.mark.slow,
|
|
]
|
|
|
|
class LocalExecutor:
|
|
"""`SshExecutor.run`'s contract, run against the local shell.
|
|
|
|
On timeout it kills only the outer shell -- exactly what an SSH channel
|
|
teardown does -- so a `setsid`-detached child survives, which is the whole
|
|
behaviour a background job depends on.
|
|
"""
|
|
|
|
def __init__(self, project_dir: str) -> None:
|
|
self.project_dir = project_dir
|
|
|
|
async def run(self, request: ExecRequest) -> ExecResult:
|
|
directory = request.cwd or self.project_dir
|
|
command = request.command
|
|
if directory:
|
|
command = f"cd {ssh_service._quote(directory)} && {command}"
|
|
started = time.monotonic()
|
|
proc = await asyncio.create_subprocess_shell(
|
|
command,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.STDOUT,
|
|
stdin=asyncio.subprocess.DEVNULL,
|
|
)
|
|
try:
|
|
out, _ = await asyncio.wait_for(proc.communicate(), timeout=request.timeout)
|
|
except TimeoutError:
|
|
proc.kill()
|
|
await proc.wait()
|
|
return ExecResult(
|
|
exit_status=-1,
|
|
output=f"The command was still running after {request.timeout:g}s.",
|
|
timed_out=True,
|
|
duration_ms=int((time.monotonic() - started) * 1000),
|
|
)
|
|
output, truncated = clean_output(out.decode("utf-8", "replace"), limit=request.max_bytes)
|
|
return ExecResult(
|
|
exit_status=proc.returncode if proc.returncode is not None else -1,
|
|
output=output,
|
|
truncated=truncated,
|
|
duration_ms=int((time.monotonic() - started) * 1000),
|
|
)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clear_jobs():
|
|
jobs.clear()
|
|
yield
|
|
jobs.clear()
|
|
|
|
|
|
def _agent(tmp_path, **over) -> AgentContext:
|
|
fields = {
|
|
"chat_id": "chat0001",
|
|
"label": "Box",
|
|
"project_dir": str(tmp_path),
|
|
"mode": policy.MODE_AUTO,
|
|
"max_output": 65536,
|
|
"timeout": 60.0,
|
|
"max_timeout": 600.0,
|
|
"background": True,
|
|
"background_on_timeout": True,
|
|
"background_notify": True,
|
|
}
|
|
fields.update(over)
|
|
ctx = AgentContext(**fields)
|
|
ctx.executor = lambda: LocalExecutor(fields["project_dir"]) # type: ignore[method-assign]
|
|
return ctx
|
|
|
|
|
|
def _context(agent):
|
|
return tools_service.ToolContext(owner_id="u", agent=agent)
|
|
|
|
|
|
# --- The wrappers, as strings --------------------------------------------------
|
|
def test_the_command_is_never_in_a_quoted_context():
|
|
"""git commit -m 'fix' would shatter sh -c '<cmd>'. It is base64'd instead."""
|
|
import base64
|
|
|
|
cmd = "git commit -m 'fix: it'"
|
|
wrapper = jobs.launch_and_wait_command("chatx", "abc123abc123", cmd, 4096)
|
|
assert cmd not in wrapper, "the command leaked into the wrapper as shell text"
|
|
blob = wrapper.split("printf %s '", 1)[1].split("'", 1)[0]
|
|
inner = base64.b64decode(blob).decode()
|
|
assert cmd in inner, "the command was lost in the base64 round-trip"
|
|
|
|
|
|
def test_every_wrapper_is_valid_shell():
|
|
"""`sh -n` parses without executing.
|
|
|
|
The one that would have caught it: `{log}` for `{logf}` formatted the module
|
|
logger into the launch-and-wait wrapper, and `<Logger … (WARNING)>` is shell
|
|
syntax. The error landed after the sentinel, where nothing reads it, so the
|
|
command still worked and the cleanup silently never ran.
|
|
"""
|
|
import subprocess
|
|
|
|
wrappers = [
|
|
jobs.launch_command("chatx", "abc123abc123", "echo hi"),
|
|
jobs.launch_and_wait_command("chatx", "abc123abc123", "echo hi", 4096),
|
|
jobs.read_command("chatx", "abc123abc123", 4096),
|
|
jobs.stop_command("chatx", "abc123abc123"),
|
|
jobs.cleanup_command("chatx", "abc123abc123"),
|
|
]
|
|
for wrapper in wrappers:
|
|
done = subprocess.run(
|
|
["sh", "-n"], input=wrapper, capture_output=True, text=True, check=False
|
|
)
|
|
assert done.returncode == 0, f"not valid shell:\n{wrapper}\n{done.stderr}"
|
|
|
|
|
|
def test_launch_and_wait_removes_every_file_it_made():
|
|
"""Its last line is the only cleanup on the fast path -- nothing calls
|
|
`_cleanup` when a command finishes in time."""
|
|
wrapper = jobs.launch_and_wait_command("chatx", "abc123abc123", "echo hi", 4096)
|
|
removal = next(line for line in wrapper.splitlines() if line.startswith("rm -f"))
|
|
|
|
for extension in ("sh", "pid", "log", "exit"):
|
|
assert jobs._file("chatx", "abc123abc123", extension) in removal, extension
|
|
assert "<Logger" not in wrapper
|
|
|
|
|
|
def test_parse_reads_the_last_sentinel():
|
|
out = ("line one\n__LEMBAS_jobjobjob01__:0\nmore\n__LEMBAS_jobjobjob01__:0")
|
|
done = jobs.parse_completed(out, "jobjobjob01")
|
|
assert done.exit_status == 0
|
|
|
|
|
|
def test_valid_id_refuses_a_path():
|
|
assert jobs.valid_id("deadbeef0000")
|
|
assert not jobs.valid_id("../../etc/passwd")
|
|
assert not jobs.valid_id("")
|
|
|
|
|
|
# --- Real shell: launch-and-wait -----------------------------------------------
|
|
async def test_a_fast_command_completes_like_a_foreground_one(tmp_path):
|
|
agent = _agent(tmp_path)
|
|
outcome = await _run_shell(_context(agent), {"command": "echo hello"})
|
|
|
|
assert "hello" in outcome.content
|
|
assert outcome.event["status"] == "ok"
|
|
assert not jobs.for_chat(agent.chat_id), "a finished command is not a job"
|
|
|
|
|
|
async def test_a_fast_command_leaves_nothing_behind(tmp_path):
|
|
"""With background on, *every* command goes through the wrapper, so a
|
|
cleanup that does not run is four files per command on somebody's machine --
|
|
including the log, which holds everything the command printed."""
|
|
import os
|
|
import pathlib
|
|
import uuid
|
|
|
|
# Its own chat id, so the directory is exclusively this run's. Job files are
|
|
# namespaced by chat, so that is isolation by construction rather than by
|
|
# tidying up after a previous run -- which is what a shared id would need,
|
|
# and would quietly pass the moment the tidying broke.
|
|
chat_id = uuid.uuid4().hex[:12]
|
|
agent = _agent(tmp_path, chat_id=chat_id)
|
|
await _run_shell(_context(agent), {"command": "echo hello"})
|
|
|
|
root = pathlib.Path(os.environ.get("TMPDIR", "/tmp")) / "lembas-jobs" / chat_id
|
|
left = sorted(p.name for p in root.iterdir()) if root.exists() else []
|
|
assert left == [], f"left behind: {left}"
|
|
|
|
|
|
async def test_a_nonzero_exit_is_read_from_the_exit_file_not_the_wrapper(tmp_path):
|
|
"""The wrapper's own status is ~0 from its trailing rm; the command's real
|
|
status is in the exit-file."""
|
|
agent = _agent(tmp_path)
|
|
outcome = await _run_shell(_context(agent), {"command": "sh -c 'exit 3'"})
|
|
|
|
assert "exited 3" in outcome.content
|
|
assert outcome.event["status"] == "error"
|
|
|
|
|
|
async def test_a_command_with_single_quotes_runs(tmp_path):
|
|
agent = _agent(tmp_path)
|
|
outcome = await _run_shell(_context(agent), {"command": "echo 'a'\\''b'"})
|
|
|
|
assert "a'b" in outcome.content
|
|
|
|
|
|
async def test_output_is_captured(tmp_path):
|
|
agent = _agent(tmp_path)
|
|
outcome = await _run_shell(_context(agent), {"command": "printf 'one\\ntwo\\n'"})
|
|
assert "one" in outcome.content and "two" in outcome.content
|
|
|
|
|
|
# --- Real shell: auto-background on timeout -------------------------------------
|
|
async def test_a_slow_command_is_kept_running_when_it_times_out(tmp_path):
|
|
"""The apt-install case. It is not killed; it becomes a job that finishes
|
|
on its own, and its result is readable afterwards."""
|
|
agent = _agent(tmp_path)
|
|
outcome = await _run_shell(
|
|
_context(agent), {"command": "sleep 2; echo done-late", "timeout": 1}
|
|
)
|
|
|
|
assert "background" in outcome.content.lower()
|
|
running = jobs.for_chat(agent.chat_id)
|
|
assert len(running) == 1
|
|
job_id = running[0].id
|
|
|
|
# The detached child survived the wait being cut off; give it time to finish.
|
|
for _ in range(40):
|
|
reading = await jobs.read(agent, job_id)
|
|
if reading.status == "done":
|
|
break
|
|
await asyncio.sleep(0.2)
|
|
assert reading.status == "done"
|
|
assert "done-late" in reading.body
|
|
|
|
|
|
async def test_off_timeout_keeps_the_hard_stop(tmp_path):
|
|
"""With the auto-convert off, a timed-out command is killed as before and no
|
|
job is left running."""
|
|
agent = _agent(tmp_path, background_on_timeout=False)
|
|
outcome = await _run_shell(
|
|
_context(agent), {"command": "sleep 2; echo late", "timeout": 1}
|
|
)
|
|
|
|
assert "stopped after" in outcome.content
|
|
assert not jobs.for_chat(agent.chat_id)
|
|
|
|
|
|
async def test_feature_off_is_the_plain_path(tmp_path):
|
|
"""No wrapper, no job, byte-for-byte the old wording."""
|
|
agent = _agent(tmp_path, background=False)
|
|
outcome = await _run_shell(_context(agent), {"command": "echo hi"})
|
|
|
|
assert "hi" in outcome.content
|
|
assert not jobs.for_chat(agent.chat_id)
|
|
|
|
|
|
# --- Explicit background=true --------------------------------------------------
|
|
async def test_background_true_returns_at_once_with_a_job(tmp_path):
|
|
agent = _agent(tmp_path)
|
|
started = time.monotonic()
|
|
outcome = await _run_shell(
|
|
_context(agent), {"command": "sleep 5", "background": True}
|
|
)
|
|
assert time.monotonic() - started < 3, "it should not wait for the command"
|
|
|
|
running = jobs.for_chat(agent.chat_id)
|
|
assert len(running) == 1
|
|
assert "job " in outcome.content
|
|
await jobs.stop(agent, running[0].id)
|
|
|
|
|
|
# --- The job tools -------------------------------------------------------------
|
|
async def test_job_output_reads_a_running_job(tmp_path):
|
|
from lembas.services.agent.tools import _run_job_output
|
|
|
|
agent = _agent(tmp_path)
|
|
job = await jobs.launch(agent, "sleep 5")
|
|
outcome = await _run_job_output(_context(agent), {"id": job.id})
|
|
assert "running" in outcome.content.lower()
|
|
await jobs.stop(agent, job.id)
|
|
|
|
|
|
async def test_job_stop_ends_it(tmp_path):
|
|
from lembas.services.agent.tools import _run_job_output, _run_job_stop
|
|
|
|
agent = _agent(tmp_path)
|
|
job = await jobs.launch(agent, "sleep 30")
|
|
stopped = await _run_job_stop(_context(agent), {"id": job.id})
|
|
assert "Stopped" in stopped.content
|
|
|
|
reading = await _run_job_output(_context(agent), {"id": job.id})
|
|
assert "running" not in reading.content.lower()
|
|
|
|
|
|
async def test_job_output_refuses_a_bad_id(tmp_path):
|
|
from lembas.services.agent.tools import _run_job_output
|
|
|
|
agent = _agent(tmp_path)
|
|
outcome = await _run_job_output(_context(agent), {"id": "../../etc/passwd"})
|
|
assert outcome.event["status"] == "error"
|
|
|
|
|
|
async def test_a_job_is_only_visible_to_its_own_chat(tmp_path):
|
|
"""for_chat is keyed on chat id; the file paths are too, so a model in
|
|
another chat cannot even name it."""
|
|
agent_a = _agent(tmp_path, chat_id="chataaaa")
|
|
await jobs.launch(agent_a, "sleep 5")
|
|
assert jobs.for_chat("chatbbbb") == []
|
|
for job in jobs.for_chat("chataaaa"):
|
|
await jobs.stop(agent_a, job.id)
|
|
|
|
|
|
# --- Gating --------------------------------------------------------------------
|
|
def test_job_tools_appear_only_when_enabled(tmp_path):
|
|
from lembas.services.agent.tools import tool_defs
|
|
|
|
on = {t.name for t in tool_defs(_agent(tmp_path, background=True))}
|
|
off = {t.name for t in tool_defs(_agent(tmp_path, background=False))}
|
|
|
|
assert {"job_output", "job_list", "job_stop"} <= on
|
|
assert not ({"job_output", "job_list", "job_stop"} & off)
|
|
|
|
|
|
def test_background_param_appears_only_when_enabled(tmp_path):
|
|
from lembas.services.agent.tools import tool_defs
|
|
|
|
def shell(agent):
|
|
return next(t for t in tool_defs(agent) if t.name == "shell_run")
|
|
|
|
on = shell(_agent(tmp_path, background=True)).parameters["properties"]
|
|
off = shell(_agent(tmp_path, background=False)).parameters["properties"]
|
|
assert "background" in on
|
|
assert "background" not in off
|
|
|
|
|
|
# --- Waking the model when a job finishes --------------------------------------
|
|
def _chat(db, user_id) -> str:
|
|
from lembas.db.models import Chat
|
|
|
|
chat = Chat(user_id=user_id, model_id="m")
|
|
db.add(chat)
|
|
db.commit()
|
|
return chat.id
|
|
|
|
|
|
def test_the_completion_names_itself_a_machine_event():
|
|
text = jobs._completion_text("abc123abc123", "apt-get install -y x", "done", 0, "ok\n")
|
|
assert "machine event" in text
|
|
assert "[job abc123abc123]" in text
|
|
assert "apt-get install" in text
|
|
assert "finished successfully" in text.lower()
|
|
|
|
|
|
def test_a_nonzero_completion_reports_the_code():
|
|
text = jobs._completion_text("j", "build", "done", 2, "")
|
|
assert "exited 2" in text
|
|
|
|
|
|
def test_backticks_in_output_cannot_close_the_fence():
|
|
text = jobs._completion_text("j", "c", "done", 0, "see ```code``` here")
|
|
assert "```code```" not in text
|
|
|
|
|
|
async def test_an_idle_chat_gets_a_fresh_reply(db, user_id, registered, monkeypatch):
|
|
from lembas.services import generation as generation_service
|
|
|
|
chat_id = _chat(db, user_id)
|
|
monkeypatch.setattr(generation_service, "running_for", lambda _c: None)
|
|
started: list[str] = []
|
|
monkeypatch.setattr(generation_service, "ensure", lambda c, m: started.append(m))
|
|
|
|
await jobs.wake(chat_id, "abc123abc123", "sleep 1", "done", 0, "hi")
|
|
|
|
from lembas.db.models import Message
|
|
|
|
db.expire_all()
|
|
msgs = db.query(Message).filter(Message.chat_id == chat_id).all()
|
|
users = [m for m in msgs if m.role == "user"]
|
|
assert len(users) == 1
|
|
assert users[0].queued is False, "an idle chat's completion is delivered, not queued"
|
|
assert any(m.role == "assistant" and not m.complete for m in msgs)
|
|
assert len(started) == 1, "a reply was started"
|
|
|
|
|
|
async def test_a_busy_chat_gets_a_queued_turn_and_no_new_reply(
|
|
db, user_id, registered, monkeypatch
|
|
):
|
|
from lembas.services import generation as generation_service
|
|
|
|
chat_id = _chat(db, user_id)
|
|
monkeypatch.setattr(generation_service, "running_for", lambda _c: object())
|
|
started: list[str] = []
|
|
monkeypatch.setattr(generation_service, "ensure", lambda c, m: started.append(m))
|
|
|
|
await jobs.wake(chat_id, "abc123abc123", "sleep 1", "done", 0, "hi")
|
|
|
|
from lembas.db.models import Message
|
|
|
|
db.expire_all()
|
|
users = [
|
|
m for m in db.query(Message).filter(Message.chat_id == chat_id).all() if m.role == "user"
|
|
]
|
|
assert len(users) == 1
|
|
assert users[0].queued is True, "a running reply's _inject/_drain will deliver it"
|
|
assert started == [], "no second reply for a chat already writing one"
|
|
|
|
|
|
async def test_two_jobs_finishing_at_once_start_one_reply(db, user_id, registered, monkeypatch):
|
|
"""The per-chat lock. Without it, both wakes would each start a generation."""
|
|
from lembas.services import generation as generation_service
|
|
|
|
chat_id = _chat(db, user_id)
|
|
live = {"running": False}
|
|
monkeypatch.setattr(
|
|
generation_service, "running_for", lambda _c: object() if live["running"] else None
|
|
)
|
|
started: list[str] = []
|
|
|
|
def _ensure(c, m):
|
|
live["running"] = True # what running_for will now see
|
|
started.append(m)
|
|
|
|
monkeypatch.setattr(generation_service, "ensure", _ensure)
|
|
|
|
await asyncio.gather(
|
|
jobs.wake(chat_id, "aaaaaaaaaaaa", "a", "done", 0, "x"),
|
|
jobs.wake(chat_id, "bbbbbbbbbbbb", "b", "done", 0, "y"),
|
|
)
|
|
|
|
assert len(started) == 1, "the lock made the second wake see the first's reply"
|
|
|
|
|
|
async def test_a_completion_is_marked_as_a_machine_event(db, user_id, registered, monkeypatch):
|
|
"""The role stays `user` -- `_inject` sends a queued turn verbatim and
|
|
`build_messages` has to keep seeing a user turn -- and `machine` is what
|
|
stops the transcript claiming the reader typed it."""
|
|
from lembas.services import generation as generation_service
|
|
|
|
chat_id = _chat(db, user_id)
|
|
monkeypatch.setattr(generation_service, "running_for", lambda _c: None)
|
|
monkeypatch.setattr(generation_service, "ensure", lambda c, m: None)
|
|
|
|
await jobs.wake(chat_id, "abc123abc123", "sleep 1", "done", 0, "hi")
|
|
|
|
from lembas.db.models import Message
|
|
|
|
db.expire_all()
|
|
users = [
|
|
m for m in db.query(Message).filter(Message.chat_id == chat_id).all() if m.role == "user"
|
|
]
|
|
assert users[0].role == "user", "the wire role is load-bearing and must not move"
|
|
assert users[0].machine is True
|
|
|
|
|
|
async def test_a_queued_completion_is_marked_too(db, user_id, registered, monkeypatch):
|
|
"""The busy path writes the same row with `queued` set; it must not lose the
|
|
marking on the way, or a completion delivered by `_drain` arrives wearing the
|
|
reader's name."""
|
|
from lembas.services import generation as generation_service
|
|
|
|
chat_id = _chat(db, user_id)
|
|
monkeypatch.setattr(generation_service, "running_for", lambda _c: object())
|
|
monkeypatch.setattr(generation_service, "ensure", lambda c, m: None)
|
|
|
|
await jobs.wake(chat_id, "abc123abc123", "sleep 1", "done", 0, "hi")
|
|
|
|
from lembas.db.models import Message
|
|
|
|
db.expire_all()
|
|
users = [
|
|
m for m in db.query(Message).filter(Message.chat_id == chat_id).all() if m.role == "user"
|
|
]
|
|
assert users[0].queued is True
|
|
assert users[0].machine is True
|
|
|
|
|
|
def test_the_prompt_quotes_the_words_the_completion_actually_carries():
|
|
"""`tool.background` tells the model a completion "begins" with a particular
|
|
sentence, so that it reads one as a machine event rather than as the person
|
|
speaking. Rewording `_completion_text` breaks that instruction, in a way
|
|
nothing else here would notice -- the turn still arrives, the model just
|
|
stops being told what it is."""
|
|
from lembas.services import prompts as prompts_service
|
|
|
|
text = jobs._completion_text("abc123abc123", "pytest -q", "done", 0, "")
|
|
opening = "A background job you started has finished"
|
|
assert text.startswith(opening)
|
|
|
|
fragment = next(f for f in prompts_service.BUILTIN if f.key == "tool.background")
|
|
assert opening in fragment.default
|
|
|
|
|
|
# --- What a job looks like in the panel ----------------------------------------
|
|
@pytest.mark.parametrize(
|
|
("status", "exit_status", "expected"),
|
|
[
|
|
("running", None, "running"),
|
|
("done", 0, "ok"),
|
|
("done", 2, "failed"),
|
|
("killed", 143, "killed"),
|
|
("lost", None, "lost"),
|
|
],
|
|
)
|
|
def test_the_dot_tells_a_failure_from_a_success(status, exit_status, expected):
|
|
"""`status` is `done` for exit 0 and exit 2 alike, and the row beside the dot
|
|
already says "Finished" or "Failed, exit 2". A dot keyed on the status would
|
|
be green next to the sentence contradicting it."""
|
|
view = jobs.JobView(id="a", command="x", status=status, exit_status=exit_status)
|
|
assert view.tone == expected
|
|
|
|
|
|
def test_a_finished_job_says_how_long_it_took():
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
started = datetime(2026, 8, 5, 14, 0, tzinfo=UTC)
|
|
view = jobs.JobView(
|
|
id="a",
|
|
command="x",
|
|
status="done",
|
|
exit_status=0,
|
|
started_at=started,
|
|
finished_at=started + timedelta(seconds=252),
|
|
)
|
|
assert view.duration == "4m 12s"
|
|
|
|
|
|
def test_a_running_job_reports_no_duration():
|
|
"""Not for want of an answer. The panel is fetched when somebody opens it and
|
|
never polled, so a live figure would be frozen the instant it painted."""
|
|
from datetime import UTC, datetime
|
|
|
|
view = jobs.JobView(
|
|
id="a",
|
|
command="x",
|
|
status="running",
|
|
started_at=datetime(2026, 8, 5, 14, 0, tzinfo=UTC),
|
|
)
|
|
assert view.duration == ""
|
|
|
|
|
|
def test_a_job_with_no_row_reports_no_duration():
|
|
"""`_persist_row` is best-effort by design, so a job with no stamps is a real
|
|
case rather than a defensive one."""
|
|
assert jobs.JobView(id="a", command="x", status="done", exit_status=0).duration == ""
|
|
|
|
|
|
def test_a_duration_survives_a_stamp_read_back_from_disk():
|
|
"""SQLite stores no offset, so a row loaded from disk comes back naive while
|
|
one still in the session's identity map keeps its tzinfo -- and a job started
|
|
before a restart and finished after it has one of each. Subtracting them
|
|
without normalising raises, and it raises in the panel, not in a test."""
|
|
from datetime import UTC, datetime
|
|
|
|
view = jobs.JobView(
|
|
id="a",
|
|
command="x",
|
|
status="done",
|
|
exit_status=0,
|
|
started_at=datetime(2026, 8, 5, 14, 0), # naive, as SQLite hands it back
|
|
finished_at=datetime(2026, 8, 5, 15, 6, tzinfo=UTC),
|
|
)
|
|
assert view.duration == "1h 06m"
|
|
|
|
|
|
# --- The watcher, end to end ---------------------------------------------------
|
|
async def test_the_watcher_end_to_end(db, user_id, registered, monkeypatch, tmp_path):
|
|
from lembas.services.agent import ssh as ssh_service
|
|
|
|
chat_id = _chat(db, user_id)
|
|
|
|
class _Local(LocalExecutor):
|
|
def __init__(self, spec, project_dir):
|
|
super().__init__(str(tmp_path))
|
|
|
|
monkeypatch.setattr(ssh_service, "SshExecutor", _Local)
|
|
monkeypatch.setattr(jobs, "_poll_interval", lambda _e: 0.1)
|
|
woken: list = []
|
|
|
|
async def _wake(chat, job_id, command, status, code, body):
|
|
woken.append((status, code, body))
|
|
|
|
monkeypatch.setattr(jobs, "wake", _wake)
|
|
|
|
agent = _agent(tmp_path, chat_id=chat_id)
|
|
agent.executor = lambda: _Local(None, str(tmp_path))
|
|
job = await jobs.launch(agent, "sleep 0.5; echo finished")
|
|
jobs.start_watch(agent, job)
|
|
|
|
for _ in range(50):
|
|
if woken:
|
|
break
|
|
await asyncio.sleep(0.1)
|
|
assert woken, "the watcher never woke the model"
|
|
status, code, body = woken[0]
|
|
assert status == "done"
|
|
assert "finished" in body
|