|
@@ -204,6 +204,21 @@ class FakeSaddle:
|
|
|
self.fail_next = False # next /start drives the task to failed
|
|
self.fail_next = False # next /start drives the task to failed
|
|
|
self.start_counts = {} # task id -> /start count
|
|
self.start_counts = {} # task id -> /start count
|
|
|
self.resumed_pending = False # crash-resume flag on the next /status
|
|
self.resumed_pending = False # crash-resume flag on the next /status
|
|
|
|
|
+ self.pending_output = [] # entries drained by the next /output poll
|
|
|
|
|
+ self.output_seq = 100
|
|
|
|
|
+
|
|
|
|
|
+ def queue_output(self, entry_type, command, output):
|
|
|
|
|
+ """Queues a typed entry for the dispatcher's next /output poll."""
|
|
|
|
|
+ with self.lock:
|
|
|
|
|
+ self.pending_output.append({
|
|
|
|
|
+ "id": self.output_seq,
|
|
|
|
|
+ "command": command,
|
|
|
|
|
+ "output": output,
|
|
|
|
|
+ "exit_code": 1,
|
|
|
|
|
+ "type": entry_type,
|
|
|
|
|
+ "timestamp": "2026-09-06T00:00:00Z",
|
|
|
|
|
+ })
|
|
|
|
|
+ self.output_seq += 1
|
|
|
|
|
|
|
|
def append_chunk(self, text):
|
|
def append_chunk(self, text):
|
|
|
with self.lock:
|
|
with self.lock:
|
|
@@ -278,7 +293,10 @@ class FakeSaddle:
|
|
|
self.daemon.summary = "fake daemon finished"
|
|
self.daemon.summary = "fake daemon finished"
|
|
|
self._send(200, response)
|
|
self._send(200, response)
|
|
|
elif self.path.startswith("/output"):
|
|
elif self.path.startswith("/output"):
|
|
|
- self._send(200, [])
|
|
|
|
|
|
|
+ with self.daemon.lock:
|
|
|
|
|
+ entries = self.daemon.pending_output
|
|
|
|
|
+ self.daemon.pending_output = []
|
|
|
|
|
+ self._send(200, entries)
|
|
|
elif self.path.startswith("/log"):
|
|
elif self.path.startswith("/log"):
|
|
|
query = urllib.parse.urlparse(self.path).query
|
|
query = urllib.parse.urlparse(self.path).query
|
|
|
since = int(urllib.parse.parse_qs(query).get("since", ["0"])[0])
|
|
since = int(urllib.parse.parse_qs(query).get("since", ["0"])[0])
|
|
@@ -839,6 +857,60 @@ def main():
|
|
|
"refreshed saddle returns idle")
|
|
"refreshed saddle returns idle")
|
|
|
check(wait_task_done(auth, stuck, timeout=60) is not None,
|
|
check(wait_task_done(auth, stuck, timeout=60) is not None,
|
|
|
"requeued task resumes on the refreshed saddle and completes")
|
|
"requeued task resumes on the refreshed saddle and completes")
|
|
|
|
|
+
|
|
|
|
|
+ print("== rejected entries (non-fatal) ==")
|
|
|
|
|
+ # Ingest API: a rejected entry stores on the session and streams
|
|
|
|
|
+ # through SSE; the task is untouched.
|
|
|
|
|
+ status, _ = saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token,
|
|
|
|
|
+ {"type": "output", "session_id": session_id, "data": {
|
|
|
|
|
+ "command": "bash -n script.sh",
|
|
|
|
|
+ "output": "syntax error near line 3 (recoverable)",
|
|
|
|
|
+ "exit_code": 1,
|
|
|
|
|
+ "type": "rejected"}})
|
|
|
|
|
+ check(status == 200, "rejected entry accepted by the ingest API")
|
|
|
|
|
+ request = urllib.request.Request(WEB + "/terminal-stream/%d" % session_id)
|
|
|
|
|
+ with urllib.request.urlopen(request, timeout=15) as stream:
|
|
|
|
|
+ rejected_event = None
|
|
|
|
|
+ deadline = time.monotonic() + 8
|
|
|
|
|
+ while time.monotonic() < deadline and rejected_event is None:
|
|
|
|
|
+ line = stream.readline().decode().strip()
|
|
|
|
|
+ if not line.startswith("data: ") or line == "data: ping":
|
|
|
|
|
+ continue
|
|
|
|
|
+ event = json.loads(line[6:])
|
|
|
|
|
+ if event.get("type") == "rejected":
|
|
|
|
|
+ rejected_event = event
|
|
|
|
|
+ check(rejected_event is not None
|
|
|
|
|
+ and "syntax error near line 3" in (rejected_event.get("output") or ""),
|
|
|
|
|
+ "terminal stream carries rejected entries with their reason")
|
|
|
|
|
+
|
|
|
|
|
+ # The page template renders them with the distinct non-fatal style.
|
|
|
|
|
+ status, page = http("GET", "/tasks/%d/terminal" % stuck, timeout=15)
|
|
|
|
|
+ check(status == 200 and "term-rejected" in page,
|
|
|
|
|
+ "terminal page renders rejected entries in their own style")
|
|
|
|
|
+
|
|
|
|
|
+ # Dispatch path: a rejected entry from /output flows into the audit
|
|
|
|
|
+ # trail without failing (or finishing) the task.
|
|
|
|
|
+ fake.queue_output("rejected", "echo 'too long' | saddle",
|
|
|
|
|
+ "max_tokens exceeded (recoverable)")
|
|
|
|
|
+ act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
|
|
|
|
|
+ "project_id": "1",
|
|
|
|
|
+ "title": "Rejected ride",
|
|
|
|
|
+ "priority": "2",
|
|
|
|
|
+ })
|
|
|
|
|
+ rejected_task = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
|
|
|
|
|
+ if t["title"] == "Rejected ride"][0])
|
|
|
|
|
+ check(wait_task_done(auth, rejected_task, timeout=60) is not None,
|
|
|
|
|
+ "task completes normally past rejected entries")
|
|
|
|
|
+ conn = db()
|
|
|
|
|
+ stored = conn.execute(
|
|
|
|
|
+ "select count(*) from session_commands sc join sessions s on s.id = sc.session_id"
|
|
|
|
|
+ " where s.task_id = ? and sc.type = 'rejected'", (rejected_task,)).fetchone()[0]
|
|
|
|
|
+ conn.close()
|
|
|
|
|
+ check(stored >= 1,
|
|
|
|
|
+ "rejected entries stored in the audit trail (%d rows)" % stored)
|
|
|
|
|
+ check(task_state(auth, rejected_task)["status"] == "done"
|
|
|
|
|
+ and "fake daemon finished" in (task_state(auth, rejected_task).get("summary") or ""),
|
|
|
|
|
+ "rejected entries never fail the task")
|
|
|
finally:
|
|
finally:
|
|
|
app.terminate()
|
|
app.terminate()
|
|
|
try:
|
|
try:
|