jj.wang
31850c4a81
One tick polls in-flight runs, posts results, classifies the first line and moves state/assignee; failures get a structured note with transient retry and backoff, timeouts are killed and handed over, resume failures and long threads rebuild with a handoff summary. Then the lowest Todo items are dispatched within terminal concurrency and serialize groups, and a heartbeat is written. CLI: tick, new, show, comment, state, assign, smoke, check-config.
80 lines
3.8 KiB
Python
80 lines
3.8 KiB
Python
import unittest
|
|
|
|
from agentd.core import classify_failure, classify_first_line, parse_budget
|
|
from agentd.executors import ExecResult
|
|
from agentd.executors.claude_code import parse_output
|
|
from agentd.plane import brief_to_html, html_to_text, text_to_html
|
|
|
|
|
|
class FirstLineTest(unittest.TestCase):
|
|
def test_tags(self):
|
|
self.assertEqual(classify_first_line("[review] done\nmore"), "review")
|
|
self.assertEqual(classify_first_line("\n\n [Question] which one?"), "question")
|
|
self.assertEqual(classify_first_line("**[blocked]** no device"), "blocked")
|
|
self.assertEqual(classify_first_line("[progress]"), "progress")
|
|
|
|
def test_untagged(self):
|
|
self.assertIsNone(classify_first_line("done, see [review] below"))
|
|
self.assertIsNone(classify_first_line("[done] finished"))
|
|
self.assertIsNone(classify_first_line(""))
|
|
self.assertIsNone(classify_first_line(None))
|
|
|
|
|
|
class HtmlTest(unittest.TestCase):
|
|
def test_html_to_text(self):
|
|
self.assertEqual(html_to_text("<p>a & b</p><p>c<br/>d</p>"), "a & b\nc\nd")
|
|
self.assertEqual(html_to_text("<ul><li>one</li><li>two</li></ul>"), "- one\n- two")
|
|
self.assertEqual(html_to_text("<pre>x < y\nz</pre>"), "x < y\nz")
|
|
self.assertEqual(html_to_text(None), "")
|
|
self.assertEqual(html_to_text("<p>a</p>\n\n\n\n<p>b</p>"), "a\n\nb")
|
|
|
|
def test_text_to_html_escapes(self):
|
|
self.assertEqual(text_to_html("a<b>\nc", pre="<x>"), "<p>a<b><br>c</p><pre><x></pre>")
|
|
self.assertEqual(html_to_text(text_to_html("[worker] hi & bye")), "[worker] hi & bye")
|
|
|
|
def test_brief(self):
|
|
self.assertEqual(brief_to_html("<p>x</p>"), "<p>x</p>")
|
|
self.assertEqual(brief_to_html("a < b"), "<pre>a < b</pre>")
|
|
|
|
|
|
class FailureTest(unittest.TestCase):
|
|
def kind(self, tail, exit_code=1, timed_out=False, cfg=None):
|
|
return classify_failure(ExecResult("", exit_code, None, [], tail), timed_out, cfg or {})
|
|
|
|
def test_transient_patterns(self):
|
|
for tail in ("request timed out", "Connection reset by peer", "connection refused",
|
|
"kex_exchange_identification: banner exchange: Connection to UNKNOWN port 65535: invalid format",
|
|
"HTTP 429 Too Many Requests", "unexpected status 502 Bad Gateway", "503 Service Unavailable",
|
|
"stream disconnected before completion"):
|
|
self.assertEqual(self.kind(tail), ("crash", True), tail)
|
|
|
|
def test_not_transient(self):
|
|
self.assertEqual(self.kind("SyntaxError: invalid syntax"), ("crash", False))
|
|
self.assertEqual(self.kind("", exit_code=0), ("logic", False))
|
|
|
|
def test_timeout_and_env(self):
|
|
self.assertEqual(self.kind("whatever", timed_out=True), ("timeout", False))
|
|
self.assertEqual(self.kind("window lost focus", cfg={"env_patterns": ["lost focus"]}), ("env", False))
|
|
|
|
def test_custom_transient_patterns_replace_defaults(self):
|
|
self.assertEqual(self.kind("timed out", cfg={"transient_patterns": ["flaky"]}), ("crash", False))
|
|
self.assertEqual(self.kind("flaky relay", cfg={"transient_patterns": ["flaky"]}), ("crash", True))
|
|
|
|
def test_budget(self):
|
|
self.assertEqual(parse_budget("budget:2h"), 2)
|
|
self.assertEqual(parse_budget("budget:30m"), 0.5)
|
|
self.assertEqual(parse_budget("budget:1.5"), 1.5)
|
|
self.assertIsNone(parse_budget("budget:soon"))
|
|
|
|
|
|
class ClaudeOutputTest(unittest.TestCase):
|
|
def test_parse(self):
|
|
obj = parse_output('{"type":"result","subtype":"success","is_error":false,"result":"[review] ok","session_id":"s1"}')
|
|
self.assertEqual((obj["result"], obj["session_id"]), ("[review] ok", "s1"))
|
|
self.assertEqual(parse_output('noise\n{"result":"x","session_id":"s2"}')["session_id"], "s2")
|
|
self.assertEqual(parse_output(""), {})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|