|
|
@@ -220,6 +220,9 @@ class FakeSaddle:
|
|
|
self.scripted_log = None # fixed chunk set served by /log
|
|
|
self.scripted_next = 0 # the daemon-style next for it
|
|
|
self.scripted_corpus = None # chunks "produced" when a task starts
|
|
|
+ self.race_texts = None # chunks appended DURING /log response
|
|
|
+ # construction (the daemon-side bug)
|
|
|
+ self.answer_bodies = [] # recorded POST /answer payloads
|
|
|
self.abort_failures = 0 # /abort requests failing (500) while > 0
|
|
|
self.abort_log = [] # task ids aborted via POST /abort
|
|
|
self.running_task = 0 # task id the daemon believes it runs
|
|
|
@@ -346,6 +349,10 @@ class FakeSaddle:
|
|
|
self.daemon.scripted_next = self.daemon.scripted_corpus[-1]["seq"] + 1
|
|
|
self._send(200, {})
|
|
|
elif self.path == "/answer":
|
|
|
+ try:
|
|
|
+ self.daemon.answer_bodies.append(json.loads(raw))
|
|
|
+ except ValueError:
|
|
|
+ pass
|
|
|
self._send(200, {})
|
|
|
elif self.path == "/abort":
|
|
|
if self.daemon.abort_failures > 0:
|
|
|
@@ -421,10 +428,27 @@ class FakeSaddle:
|
|
|
return
|
|
|
query = urllib.parse.urlparse(self.path).query
|
|
|
since = int(urllib.parse.parse_qs(query).get("since", ["0"])[0])
|
|
|
+ # The daemon-side snapshot race: the served array is
|
|
|
+ # snapshotted, chunks are appended, and `next` is computed
|
|
|
+ # AFTER the appends — the response's next skips chunks the
|
|
|
+ # array never carried (the live run's 33 lost chunks).
|
|
|
+ raced = False
|
|
|
+ if since > 0 and self.daemon.race_texts:
|
|
|
+ raced = True
|
|
|
+ with self.daemon.lock:
|
|
|
+ next_seq = self.daemon.scripted_next
|
|
|
+ for text in self.daemon.race_texts:
|
|
|
+ self.daemon.scripted_log.append(
|
|
|
+ {"seq": next_seq, "text": text})
|
|
|
+ next_seq += 1
|
|
|
+ self.daemon.scripted_next = next_seq
|
|
|
+ self.daemon.race_texts = None
|
|
|
# Simulate the live PTY: each poll produces one more chunk
|
|
|
# (before slicing, so the poller actually sees it).
|
|
|
self.daemon.append_chunk("stream-line-%d\r\n" % self.daemon.next_seq)
|
|
|
chunks, nxt = self.daemon.log_since(since)
|
|
|
+ if raced:
|
|
|
+ nxt = self.daemon.scripted_next
|
|
|
self._send(200, {"chunks": chunks, "next": nxt})
|
|
|
else:
|
|
|
self._send(404, {"error": "no such route"})
|
|
|
@@ -2320,6 +2344,107 @@ def main():
|
|
|
"identity task completes")
|
|
|
|
|
|
|
|
|
+
|
|
|
+ print("== R10: branch_advanced + gap forensics ==")
|
|
|
+ # --- branch_advanced on the /poll verdict channel ---
|
|
|
+ fake.state = "working"
|
|
|
+ fake.hold_state = True
|
|
|
+ conn = db()
|
|
|
+ conn.execute("insert into tasks (project_id, title, status, priority,"
|
|
|
+ " created_at) values (1, 'Poll grandparent', 'working', 5,"
|
|
|
+ " strftime('%s','now','-300 seconds'))")
|
|
|
+ poll_grand = conn.execute("select max(id) from tasks").fetchone()[0]
|
|
|
+ conn.execute("insert into tasks (project_id, title, status, priority,"
|
|
|
+ " parent_task_id, created_at) values (1, 'Poll answer parent',"
|
|
|
+ " 'blocked', 5, ?, strftime('%s','now','-120 seconds'))",
|
|
|
+ (poll_grand,))
|
|
|
+ poll_parent = conn.execute("select max(id) from tasks").fetchone()[0]
|
|
|
+ conn.execute("insert into sessions (task_id, saddle_id, started_at)"
|
|
|
+ " values (?, ?, strftime('%s','now','-120 seconds'))",
|
|
|
+ (poll_parent, fake_id))
|
|
|
+ poll_session = conn.execute("select max(id) from sessions").fetchone()[0]
|
|
|
+ conn.execute("insert into task_blocks (task_id, question, type, answer,"
|
|
|
+ " answered_at, created_at) values (?, 'Which DB?', 'question',"
|
|
|
+ " 'SQLite', strftime('%s','now','-60 seconds'),"
|
|
|
+ " strftime('%s','now','-100 seconds'))", (poll_parent,))
|
|
|
+ conn.execute("update saddles set status = 'busy', current_task_id = ?,"
|
|
|
+ " allocated_at = strftime('%s','now') where name = 'fake-saddle'",
|
|
|
+ (poll_parent,))
|
|
|
+ conn.commit()
|
|
|
+ conn.close()
|
|
|
+ status, body = saddle_api("GET", "/api/saddle/%d/poll" % fake_id, fake_token)
|
|
|
+ check(status == 200 and len(body.get("answers", [])) == 1
|
|
|
+ and body["answers"][0].get("branch_advanced") is False,
|
|
|
+ "/poll verdicts carry branch_advanced=false with no siblings")
|
|
|
+
|
|
|
+ # A sibling under the same parent completed after the park marker.
|
|
|
+ conn = db()
|
|
|
+ conn.execute("insert into tasks (project_id, title, status, priority,"
|
|
|
+ " parent_task_id, summary, completed_at, created_at)"
|
|
|
+ " values (1, 'Poll sibling', 'done', 5, ?, 'sibling finished',"
|
|
|
+ " strftime('%s','now'), strftime('%s','now','-200 seconds'))",
|
|
|
+ (poll_grand,))
|
|
|
+ conn.execute("insert into task_blocks (task_id, question, type, answer,"
|
|
|
+ " answered_at, created_at) values (?, 'Second question?',"
|
|
|
+ " 'question', 'Yes', strftime('%s','now'),"
|
|
|
+ " strftime('%s','now','-90 seconds'))", (poll_parent,))
|
|
|
+ conn.commit()
|
|
|
+ conn.close()
|
|
|
+ status, body = saddle_api("GET", "/api/saddle/%d/poll" % fake_id, fake_token)
|
|
|
+ check(status == 200 and len(body.get("answers", [])) == 1
|
|
|
+ and body["answers"][0].get("branch_advanced") is True,
|
|
|
+ "sibling completion after park flips branch_advanced to true")
|
|
|
+
|
|
|
+ # Release the staged task: cancel it so the held-working poll loop
|
|
|
+ # does not carry it into the race subsection.
|
|
|
+ act(auth, task_state(auth, poll_parent)["cancel"], {})
|
|
|
+ fake.hold_state = False
|
|
|
+ fake.state = "idle"
|
|
|
+
|
|
|
+ # --- the /log snapshot race loses nothing ---
|
|
|
+ fake.state = "working"
|
|
|
+ fake.hold_state = True
|
|
|
+ fake.scripted_corpus = [
|
|
|
+ {"seq": 1, "text": "prompt-one\r\n"},
|
|
|
+ {"seq": 2, "text": "echo a\r\n"},
|
|
|
+ {"seq": 3, "text": "a\r\n"},
|
|
|
+ {"seq": 4, "text": "prompt-two\r\n"},
|
|
|
+ ]
|
|
|
+ fake.scripted_log = []
|
|
|
+ fake.scripted_next = 1
|
|
|
+ fake.race_texts = ["raced-five\r\n", "raced-six\r\n"]
|
|
|
+ act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
|
|
|
+ "project_id": "1",
|
|
|
+ "title": "Race forensics task",
|
|
|
+ "priority": "2",
|
|
|
+ })
|
|
|
+ race_task = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
|
|
|
+ if t["title"] == "Race forensics task"][0])
|
|
|
+
|
|
|
+ def race_seqs():
|
|
|
+ conn = db()
|
|
|
+ try:
|
|
|
+ return [r[0] for r in conn.execute(
|
|
|
+ "select seq from session_raw_log where session_id in"
|
|
|
+ " (select id from sessions where task_id = ?)"
|
|
|
+ " order by seq", (race_task,))]
|
|
|
+ finally:
|
|
|
+ conn.close()
|
|
|
+
|
|
|
+ check(wait_for(lambda: race_seqs() == [1, 2, 3, 4, 5, 6] or False,
|
|
|
+ timeout=40, what="race chunks all stored") is not None,
|
|
|
+ "the snapshot race loses nothing: complete 1..6, no gaps (%s)"
|
|
|
+ % race_seqs())
|
|
|
+ time.sleep(7)
|
|
|
+ check(race_seqs() == [1, 2, 3, 4, 5, 6],
|
|
|
+ "race chunks stored exactly once")
|
|
|
+ fake.hold_state = False
|
|
|
+ fake.state = "idle"
|
|
|
+ fake.scripted_log = None
|
|
|
+ fake.scripted_corpus = None
|
|
|
+ check(wait_task_done(auth, race_task, timeout=60) is not None,
|
|
|
+ "race forensics task completes")
|
|
|
+
|
|
|
# The daemon vanishing mid-task: three refused polls → offline,
|
|
|
# task requeued with a bot comment, UI reflects.
|
|
|
act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
|