"""Offline checks for the reference lab. No network.""" from __future__ import annotations import json import unittest from northline.handoff import allow, make_handoff, validate_handoff from northline.jobs import claim, connect, enqueue, finish from northline.loop import run_agent from northline.orders import REFUND_CEILING_CENTS, lookup_order, refund_order from northline.retrieve import RRF_K, Index from northline.scripted import looping_agent, order_agent from northline.trace import Tracer, redact class LoopTests(unittest.TestCase): def test_lookup_then_stop(self): tools = {"lookup_order": lookup_order} out = run_agent("Where is order 1042?", order_agent, tools) self.assertIsNone(out["error"]) self.assertEqual(out["steps"], 2) self.assertIn("shipped", out["final"]) self.assertIn("Parcel North", out["final"]) def test_missing_order(self): out = run_agent("Where is order 9999?", order_agent, {"lookup_order": lookup_order}) self.assertIn("not_found", out["final"]) def test_unknown_tool_is_not_executed(self): def evil(messages, tool_names): return { "content": None, "tool_calls": [{"id": "x", "name": "delete_all", "arguments": {}}], } out = run_agent("hi", evil, {"lookup_order": lookup_order}, max_steps=1) self.assertEqual(out["error"], "step_budget_exhausted") tool_msgs = [m for m in out["messages"] if m["role"] == "tool"] self.assertEqual(tool_msgs[0]["content"]["error"], "unknown_tool") def test_step_budget_stops_a_loop(self): out = run_agent("loop", looping_agent, {"lookup_order": lookup_order}, max_steps=4) self.assertEqual(out["error"], "step_budget_exhausted") self.assertEqual(out["steps"], 4) self.assertLessEqual( sum(1 for m in out["messages"] if m["role"] == "assistant"), 4, ) def test_bad_arguments_shape(self): def bad(messages, tool_names): return { "content": None, "tool_calls": [{"id": "x", "name": "lookup_order", "arguments": "1042"}], } out = run_agent("x", bad, {"lookup_order": lookup_order}, max_steps=1) tool_msgs = [m for m in out["messages"] if m["role"] == "tool"] self.assertEqual(tool_msgs[0]["content"]["error"], "arguments_must_be_object") def test_falsy_non_objects_are_not_treated_as_missing(self): def make(arguments): def bad(messages, tool_names): return { "content": None, "tool_calls": [{"id": "x", "name": "lookup_order", "arguments": arguments}], } return bad for arguments in (None, [], ""): out = run_agent("x", make(arguments), {"lookup_order": lookup_order}, max_steps=1) tool_msgs = [m for m in out["messages"] if m["role"] == "tool"] self.assertEqual(tool_msgs[0]["content"]["error"], "arguments_must_be_object", arguments) class RefundTests(unittest.TestCase): def test_ceiling_blocks_injection_sized_refund(self): ledger = [] # 500000 cents is the attack. The model may ask. The gateway refuses. out = refund_order( {"order_id": "1042", "amount_cents": 500000, "idempotency_key": "k1"}, confirmed=True, ledger=ledger, ) self.assertEqual(out["error"], "amount_blocked") self.assertEqual(ledger, []) self.assertEqual(REFUND_CEILING_CENTS, 5000) def test_unconfirmed_does_not_commit(self): ledger = [] out = refund_order( {"order_id": "1042", "amount_cents": 1000, "idempotency_key": "k2"}, confirmed=False, ledger=ledger, ) self.assertEqual(out["error"], "not_confirmed") self.assertEqual(ledger, []) def test_cancelled_order_blocked(self): ledger = [] out = refund_order( {"order_id": "1044", "amount_cents": 100, "idempotency_key": "k3"}, confirmed=True, ledger=ledger, ) self.assertEqual(out["error"], "state_blocked") def test_idempotency(self): ledger = [] args = {"order_id": "1043", "amount_cents": 1500, "idempotency_key": "same"} first = refund_order(args, confirmed=True, ledger=ledger) second = refund_order(args, confirmed=True, ledger=ledger) self.assertEqual(first["status"], "committed") self.assertEqual(second["status"], "already_committed") self.assertEqual(len(ledger), 1) class HandoffTests(unittest.TestCase): def test_round_trip_and_authority(self): h = make_handoff( goal="Explain order 1042", constraints=["do not refund"], evidence=[{"id": "order:1042", "quote": "state=shipped"}], open_questions=[], authority=["draft_reply"], done_when="A reply states the carrier and does not offer a refund.", ) self.assertTrue(allow(h, "draft_reply")) self.assertFalse(allow(h, "refund_order")) def test_missing_field(self): with self.assertRaises(ValueError): validate_handoff({"goal": "x"}) class RetrieveTests(unittest.TestCase): def setUp(self): self.idx = Index() self.idx.upsert( chunk_id="leave-1", doc_id="leave", text="Parental leave is 12 weeks for every employee.", acl=["employee", "manager"], version=1, ) self.idx.upsert( chunk_id="leave-2", doc_id="leave", text="Parental leave is 16 weeks for every employee.", acl=["employee", "manager"], version=2, ) self.idx.tombstone_doc("leave", below_version=2) self.idx.upsert( chunk_id="comp-1", doc_id="comp", text="Compensation band L5 starts at 180000.", acl=["manager"], version=1, ) def test_employee_does_not_see_compensation_or_old_leave(self): hits = self.idx.query("parental leave policy", groups=["employee"], k=5) ids = [h["chunk_id"] for h in hits] self.assertIn("leave-2", ids) self.assertNotIn("leave-1", ids) self.assertNotIn("comp-1", ids) self.assertIn("16 weeks", hits[0]["text"]) def test_manager_can_see_compensation(self): hits = self.idx.query("compensation band L5", groups=["manager"], k=5) self.assertIn("comp-1", [h["chunk_id"] for h in hits]) def test_empty_acl_rejected(self): with self.assertRaises(ValueError): self.idx.upsert( chunk_id="x", doc_id="x", text="no group", acl=[], version=1, ) def test_rrf_constant_is_the_usual_default(self): self.assertEqual(RRF_K, 60) def test_equal_rrf_scores_break_on_chunk_id(self): import northline.retrieve as retrieve real = retrieve.embed def fake(text): vec = [0.0] * retrieve.DIM # Query "alpha" and beta-doc share a vector. alpha-doc does not. # Keyword overlap is the reverse: only alpha-doc contains "alpha". if text in {"alpha", "beta-doc"}: vec[0] = 1.0 elif text == "alpha-doc": vec[1] = 1.0 return vec retrieve.embed = fake try: idx = Index() idx.upsert(chunk_id="b-chunk", doc_id="d", text="beta-doc", acl=["employee"], version=1) idx.upsert(chunk_id="a-chunk", doc_id="d", text="alpha-doc", acl=["employee"], version=1) hits = idx.query("alpha", groups=["employee"], k=5) finally: retrieve.embed = real self.assertEqual(hits[0]["score"], hits[1]["score"]) self.assertEqual([h["chunk_id"] for h in hits], ["a-chunk", "b-chunk"]) class TraceAndJobTests(unittest.TestCase): def test_email_redacted_in_trace(self): sink = [] tracer = Tracer(sink) order = lookup_order({"order_id": "1042"}) # Do not pre-redact. Tracer.span has to do it, or the test is theater. tracer.span("tool", {"name": "lookup_order", "payload": order}) line = sink[0] self.assertNotIn("ada@example.com", line) self.assertIn("[redacted]", line) self.assertEqual(json.loads(line)["payload"]["email"], "[redacted]") parsed = json.loads(line) self.assertEqual(parsed["span"], "tool") def test_job_idempotency_and_claim(self): conn = connect() first = enqueue(conn, idempotency_key="job-1", payload="{}") second = enqueue(conn, idempotency_key="job-1", payload="{}") self.assertEqual(first, second) other = enqueue(conn, idempotency_key="job-2", payload="{}") self.assertNotEqual(other, first) retried = enqueue(conn, idempotency_key="job-1", payload="{}") self.assertEqual(retried, first) job = claim(conn) self.assertEqual(job["id"], first) self.assertEqual(job["status"], "running") self.assertEqual(job["attempts"], 1) second_job = claim(conn) self.assertEqual(second_job["id"], other) self.assertIsNone(claim(conn)) finish(conn, first, "done") row = conn.execute("SELECT status FROM jobs WHERE id = ?", (first,)).fetchone() self.assertEqual(row["status"], "done") conn.close() class HttpTests(unittest.TestCase): def test_health_is_open(self): from northline.httpapi import dispatch status, body = dispatch("GET", "/health", {}, b"", token="") self.assertEqual(status, 200) self.assertTrue(body["ok"]) def test_tasks_require_token_and_auth(self): from northline.httpapi import dispatch status, body = dispatch("POST", "/v1/tasks", {}, b'{"task":"hi"}', token="") self.assertEqual(status, 500) status, body = dispatch( "POST", "/v1/tasks", {"Authorization": "Bearer no"}, b'{"task":"Where is order 1042?"}', token="yes" ) self.assertEqual(status, 401) def test_happy_path_hides_transcript(self): from northline.httpapi import dispatch status, body = dispatch( "POST", "/v1/tasks", {"Authorization": "Bearer yes"}, b'{"task":"Where is order 1042?"}', token="yes", ) self.assertEqual(status, 200) self.assertIn("shipped", body["final"]) self.assertNotIn("messages", body) self.assertNotIn("ada@example.com", json.dumps(body)) def test_oversize_body(self): from northline.httpapi import dispatch status, body = dispatch( "POST", "/v1/tasks", {"Authorization": "Bearer yes"}, b'{"task":"x"}' + b" ", token="yes", max_body=5, ) self.assertEqual(status, 413) class Week1CaseTests(unittest.TestCase): def test_fifteen_scripted_cases(self): from northline.eval_week1 import run_all from northline.scripted import order_agent rows = run_all(order_agent) self.assertEqual(len(rows), 15) failed = [(cid, errs) for cid, errs, _steps in rows if errs] self.assertEqual(failed, []) class JudgeFixtureTests(unittest.TestCase): def test_polite_wrong_fails_because_money_moved(self): from northline.judge import FIXTURES, rubric_pass by_id = {row["id"]: row for row in FIXTURES} polite = by_id["polite-wrong"] self.assertIn("refunded the full amount", polite["draft"]) self.assertFalse(polite["answers"]["insults_or_motive"]) self.assertTrue(polite["answers"]["money_moved"]) self.assertFalse(rubric_pass(polite["answers"])) self.assertTrue(rubric_pass(by_id["carrier-only"]["answers"])) class CacheTests(unittest.TestCase): def test_groups_and_version_are_part_of_the_key(self): from northline.cache import RetrievalCache cache = RetrievalCache() cache.put("parental leave", ["employee"], "hash-v1", "corpus-1", ["leave-2"]) self.assertEqual( cache.get("parental leave", ["employee"], "hash-v1", "corpus-1"), ["leave-2"], ) self.assertIsNone(cache.get("parental leave", ["manager"], "hash-v1", "corpus-1")) self.assertIsNone(cache.get("parental leave", ["employee"], "hash-v2", "corpus-1")) self.assertIsNone(cache.get("parental leave", ["employee"], "hash-v1", "corpus-2")) cache.put("parental leave", ["manager"], "hash-v1", "corpus-1", ["comp-1", "leave-2"]) self.assertEqual( cache.get("parental leave", ["employee"], "hash-v1", "corpus-1"), ["leave-2"], ) class CapstoneTests(unittest.TestCase): def test_capstone_hostile_refund(self): from northline.handoff import allow, make_handoff from northline.orders import lookup_order, refund_order from northline.retrieve import Index from northline.trace import Tracer idx = Index() idx.upsert(chunk_id="leave-1", doc_id="leave", text="Parental leave is 12 weeks for every employee.", acl=["employee", "manager"], version=1) idx.upsert(chunk_id="leave-2", doc_id="leave", text="Parental leave is 16 weeks for every employee.", acl=["employee", "manager"], version=2) idx.tombstone_doc("leave", below_version=2) idx.upsert(chunk_id="comp-1", doc_id="comp", text="Compensation band L5 starts at 180000.", acl=["manager"], version=1) hits = idx.query("parental leave", groups=["employee"], k=5) ids = [h["chunk_id"] for h in hits] self.assertIn("leave-2", ids) self.assertNotIn("comp-1", ids) self.assertNotIn("leave-1", ids) quote = hits[0]["text"] handoff = make_handoff( goal="Explain order 1042 without committing a refund", constraints=["do not refund"], evidence=[{"id": "leave-2", "quote": quote}, {"id": "order:1042", "quote": "state=shipped carrier=Parcel North"}], open_questions=[], authority=["draft_reply"], done_when="Does not claim money moved.", ) self.assertFalse(allow(handoff, "refund_order")) self.assertTrue(quote in handoff["evidence"][0]["quote"]) ledger = [] blocked = refund_order( {"order_id": "1042", "amount_cents": 500000, "idempotency_key": "capstone"}, confirmed=True, ledger=ledger, ) self.assertEqual(blocked["error"], "amount_blocked") self.assertEqual(ledger, []) forged = refund_order( {"order_id": "1042", "amount_cents": 1000, "idempotency_key": "capstone-2"}, confirmed=False, ledger=ledger, ) self.assertEqual(forged["error"], "not_confirmed") sink = [] Tracer(sink).span("tool", {"name": "lookup_order", "payload": lookup_order({"order_id": "1042"})}) self.assertNotIn("ada@example.com", sink[0]) def test_spans_for_lookup(self): import json as _json from northline.loop import run_agent from northline.orders import lookup_order from northline.scripted import order_agent from northline.trace import Tracer sink = [] tracer = Tracer(sink) out = run_agent("Where is order 1042?", order_agent, {"lookup_order": lookup_order}, trace=tracer) self.assertIn("shipped", out["final"]) spans = [_json.loads(line) for line in sink] self.assertEqual(len({s["trace_id"] for s in spans}), 1) names = [s["span"] for s in spans] self.assertEqual(names.count("model"), 2) self.assertEqual(names.count("tool"), 1) tool_spans = [s for s in spans if s["span"] == "tool"] self.assertEqual(tool_spans[0]["name"], "lookup_order") self.assertIs(tool_spans[0]["ok"], True) self.assertNotIn("payload", tool_spans[0]) self.assertNotEqual(spans[0]["span"], "tool") self.assertNotIn("ada@example.com", "".join(sink)) class HttpBudgetTests(unittest.TestCase): def test_rate_limit_does_not_call_the_model(self): from northline.httpapi import dispatch def boom(messages, tool_names): raise AssertionError("model must not run") status, body = dispatch( "POST", "/v1/tasks", {"Authorization": "Bearer yes"}, b'{"task":"Where is order 1042?"}', token="yes", complete=boom, admit=False, ) self.assertEqual(status, 429) self.assertEqual(body["error"], "rate_limited") def test_unauthorized_does_not_call_the_model(self): from northline.httpapi import dispatch def boom(messages, tool_names): raise AssertionError("model must not run") status, body = dispatch( "POST", "/v1/tasks", {"Authorization": "Bearer no"}, b'{"task":"Where is order 1042?"}', token="yes", complete=boom, ) self.assertEqual(status, 401) def test_response_includes_trace_id_not_transcript(self): from northline.httpapi import dispatch status, body = dispatch( "POST", "/v1/tasks", {"Authorization": "Bearer yes"}, b'{"task":"Where is order 1042?"}', token="yes", ) self.assertEqual(status, 200) self.assertTrue(body["trace_id"]) self.assertNotIn("messages", body) self.assertNotIn("ada@example.com", json.dumps(body)) if __name__ == "__main__": unittest.main()