About 1.8 million lines of code changed in 10 days on an internal NEXT4I project split into 1,243 tickets. I didn't read them. Nobody could. What I did was build the line they run through.
This post turns the main ideas into a small toy you can run. It's not my real pipeline (that one is wired into internal services and agent tooling), but the control logic has the same shape:
- Check the ticket before it enters the line
- Route by complexity, so easy work goes to small models
- A TDD gate that wants evidence on both sides
- Escalate on doubt, and park for a human at the end
- A cost ceiling on expensive tiers
- Tickets you can pull off the line, and a trace for everything
Python 3.10+, standard library only.
Why not one big prompt?
1.8 million changed lines is more than Claude Opus or Sonnet can hold in context. You can't ask for everything in one go.
So I think about it like a factory. Each station on a production line does one job and passes the work along. That's faster and more accurate than one station doing everything. A ticket is a station's job: one small concern, with a goal and the tests that prove it. Small scope keeps the agent focused, keeps the context small, and cuts down on hallucination. It's the S in SOLID (single responsibility), applied to tickets.
The line
factory.py:
"""A toy AI software factory line: ticket check, model routing, TDD gate, escalation."""
from dataclasses import dataclass, field
from enum import Enum
TIERS = ["small", "medium", "large"] # cheap to expensive
class Status(str, Enum):
TODO = "TODO"
DONE = "DONE"
DONE_WITH_CONCERNS = "DONE_WITH_CONCERNS" # finished, but a human must confirm evidence
FAILED = "FAILED"
PARKED = "PARKED" # off the line, waiting for a human
REMOVED = "REMOVED" # pulled from the line on purpose
@dataclass
class Ticket:
id: str
goal: str
concerns: list # what this ticket touches; must be exactly one
tests: list # test names that verify the goal
complexity: int # 1 = small, 2 = medium, 3 = large
status: Status = Status.TODO
trace: list = field(default_factory=list)
def log(self, msg):
self.trace.append(f"{self.id}: {msg}")
@dataclass
class Evidence:
red_failed: bool # did the test fail BEFORE the code was written?
green_passed: bool # does it pass AFTER?
concerns: list = field(default_factory=list)
def check_ticket(t):
"""Quality check on the work order itself, before it enters the line."""
problems = []
if not t.goal.strip():
problems.append("no goal")
if len(t.concerns) != 1:
problems.append("one ticket, one concern")
if not t.tests:
problems.append("no tests to verify the goal")
if t.complexity not in (1, 2, 3):
problems.append("complexity must be 1-3")
return problems
def gate(ev):
"""TDD gate: evidence on both sides, or it does not count."""
if not ev.red_failed:
return Status.FAILED # a test that never failed may be always-true
if not ev.green_passed:
return Status.FAILED
if ev.concerns:
return Status.DONE_WITH_CONCERNS
return Status.DONE
class Line:
def __init__(self, worker, ceiling="medium", allow_above_ceiling=False, attempts_per_tier=2):
self.worker = worker # worker(ticket, tier) -> Evidence
self.ceiling = TIERS.index(ceiling)
self.allow_above_ceiling = allow_above_ceiling
self.attempts = attempts_per_tier
def park(self, t, reason):
t.status = Status.PARKED
t.log(f"PARKED ({reason})")
return t.status
def run(self, t):
if t.status == Status.REMOVED:
return t.status
problems = check_ticket(t)
if problems:
return self.park(t, "bad ticket: " + ", ".join(problems))
tier = t.complexity - 1 # route by complexity: easy work to small models
while tier < len(TIERS):
if tier > self.ceiling and not self.allow_above_ceiling:
return self.park(t, f"needs approval for tier {TIERS[tier]}")
for attempt in range(1, self.attempts + 1):
status = gate(self.worker(t, TIERS[tier]))
t.log(f"{TIERS[tier]} attempt {attempt}: {status.value}")
if status == Status.DONE:
t.status = status
return status
if status == Status.DONE_WITH_CONCERNS:
break # retrying the same tier rarely helps; escalate
if status == Status.DONE_WITH_CONCERNS and tier == len(TIERS) - 1:
t.status = status # top tier: accept, a human confirms the evidence
return status
tier += 1
if tier < len(TIERS):
t.log(f"escalate to {TIERS[tier]}")
return self.park(t, "failed on every tier")
def run_all(self, tickets):
summary = {}
for t in tickets:
status = self.run(t) # one bad ticket never stops the line
summary[status.value] = summary.get(status.value, 0) + 1
return summary
def remove(t, reason):
t.status = Status.REMOVED
t.log(f"REMOVED ({reason})")
Walkthrough
1. Check the work order first
check_ticket runs before any model sees the ticket. No goal, more than one concern, or no tests means the ticket is parked with a reason. A bad work order costs nothing if it never reaches a model. (In my project, tickets go through their own quality check before the line. This toy only shows the idea.)
2. Route by complexity
tier = t.complexity - 1. Simple tickets start on the small tier. A strong model on an easy ticket tends to overthink it, overshoot the goal, and cost more. The production side should follow the ticket, because the design thinking already happened when the tickets were planned.
3. A gate that wants evidence on both sides
gate checks two things:
-
red_failed: the test failed before the code existed. If it never failed, it might be an always-true test, so the result isFAILED. -
green_passed: the test passes after the code.
A pass with open concerns becomes DONE_WITH_CONCERNS, not DONE. That separate status matters later.
4. Escalate, then park
-
FAILED: retry on the same tier (two attempts by default), then move up a tier. -
DONE_WITH_CONCERNSbelow the top tier: escalate right away. Retrying the same model rarely clears its own concerns. -
DONE_WITH_CONCERNSon the top tier: accept it, but a human still has to confirm the evidence. - Failed on every tier:
PARKEDfor a human.
This is the reject arm on a factory line, with one extra step: before rejecting, ask a smarter model.
5. A cost ceiling
ceiling="medium" means the line won't touch the large tier unless you pass allow_above_ceiling=True. Cost doesn't get to run away on its own. In my workflow each tier has a spending ceiling, and going past it needs my approval. A ticket that hits the ceiling is often a sign it's too big and should be split.
6. Removable tickets and a trace
remove() pulls a ticket off the line. run_all keeps going when one ticket parks, so one bad ticket never stops the line. Every decision lands in ticket.trace, which is how you find out later why something happened.
In my real line, I never edit a ticket once it exists. A change becomes a new ticket, and the old one stays as history (the O in SOLID: open for extension, closed for modification). This toy only has remove() to show the idea.
Run the demo
demo.py uses a scripted worker so the output is repeatable:
from factory import Evidence, Line, Ticket, remove
GOOD = Evidence(red_failed=True, green_passed=True)
CONCERN = Evidence(red_failed=True, green_passed=True, concerns=["load test not measured"])
ALWAYS_TRUE = Evidence(red_failed=False, green_passed=True)
# Scripted results per (ticket, tier) so the demo is repeatable
script = {
("T-1", "small"): [GOOD],
("T-2", "small"): [CONCERN],
("T-2", "medium"): [GOOD],
("T-3", "small"): [ALWAYS_TRUE, ALWAYS_TRUE],
("T-3", "medium"): [ALWAYS_TRUE, ALWAYS_TRUE],
}
def worker(ticket, tier):
return script[(ticket.id, tier)].pop(0)
tickets = [
Ticket("T-1", "Add email format check", ["validation"], ["test_email_format"], 1),
Ticket("T-2", "Add retry to payment webhook", ["payment"], ["test_webhook_retry"], 1),
Ticket("T-3", "Add audit log field", ["audit"], ["test_audit_field"], 1),
Ticket("T-4", "Login page and token refresh", ["ui", "auth"], ["test_login"], 2),
Ticket("T-5", "Rewrite search ranking", ["search"], ["test_ranking"], 3),
Ticket("T-6", "Old export endpoint", ["export"], ["test_export"], 1),
]
remove(tickets[5], "replaced by T-9")
line = Line(worker, ceiling="medium")
print(line.run_all(tickets))
for t in tickets:
print("\n".join(t.trace))
$ python demo.py
{'DONE': 2, 'PARKED': 3, 'REMOVED': 1}
T-1: small attempt 1: DONE
T-2: small attempt 1: DONE_WITH_CONCERNS
T-2: escalate to medium
T-2: medium attempt 1: DONE
T-3: small attempt 1: FAILED
T-3: small attempt 2: FAILED
T-3: escalate to medium
T-3: medium attempt 1: FAILED
T-3: medium attempt 2: FAILED
T-3: escalate to large
T-3: PARKED (needs approval for tier large)
T-4: PARKED (bad ticket: one ticket, one concern)
T-5: PARKED (needs approval for tier large)
T-6: REMOVED (replaced by T-9)
-
T-1: easy ticket, done on the small tier. -
T-2: small tier finished with a concern, escalated, done on medium. -
T-3: the test never failed first, so every attempt is rejected as possibly always-true. It climbs to the ceiling and parks. -
T-4: two concerns in one ticket. Parked before any model runs. -
T-5: needs the large tier, which is above the ceiling. Waits for approval. -
T-6: pulled off the line on purpose.
Tests
test_factory.py covers the rules above:
import unittest
from factory import Evidence, Line, Status, Ticket, gate, remove
GOOD = Evidence(True, True)
CONCERN = Evidence(True, True, ["evidence needs a human"])
ALWAYS_TRUE = Evidence(False, True)
def ticket(tid="T", concerns=("x",), tests=("test_x",), complexity=1):
return Ticket(tid, "goal", list(concerns), list(tests), complexity)
def scripted(results):
calls = []
def worker(t, tier):
calls.append(tier)
return results[tier].pop(0)
return worker, calls
class FactoryTest(unittest.TestCase):
def test_bad_ticket_never_reaches_a_model(self):
worker, calls = scripted({})
t = ticket(concerns=("ui", "auth"))
self.assertEqual(Line(worker).run(t), Status.PARKED)
self.assertEqual(calls, [])
def test_always_true_test_is_rejected(self):
self.assertEqual(gate(ALWAYS_TRUE), Status.FAILED)
def test_concerns_escalate_to_next_tier(self):
worker, calls = scripted({"medium": [CONCERN], "large": [GOOD]})
t = ticket(complexity=2)
self.assertEqual(Line(worker, ceiling="large").run(t), Status.DONE)
self.assertEqual(calls, ["medium", "large"])
def test_tier_above_ceiling_needs_approval(self):
worker, calls = scripted({})
t = ticket(complexity=3)
self.assertEqual(Line(worker, ceiling="medium").run(t), Status.PARKED)
self.assertEqual(calls, [])
self.assertIn("needs approval", t.trace[-1])
def test_failing_ticket_is_parked_and_line_keeps_running(self):
worker, _ = scripted({"small": [ALWAYS_TRUE] * 2 + [GOOD],
"medium": [ALWAYS_TRUE] * 2})
bad, good = ticket("BAD"), ticket("GOOD")
summary = Line(worker, ceiling="medium").run_all([bad, good])
self.assertEqual(bad.status, Status.PARKED)
self.assertEqual(good.status, Status.DONE)
self.assertEqual(summary, {"PARKED": 1, "DONE": 1})
def test_removed_ticket_is_skipped(self):
worker, calls = scripted({})
t = ticket()
remove(t, "replaced")
self.assertEqual(Line(worker).run(t), Status.REMOVED)
self.assertEqual(calls, [])
def test_top_tier_concerns_stay_for_a_human(self):
worker, _ = scripted({"large": [CONCERN]})
t = ticket(complexity=3)
status = Line(worker, ceiling="large").run(t)
self.assertEqual(status, Status.DONE_WITH_CONCERNS)
if __name__ == "__main__":
unittest.main()
$ python -m unittest -v
test_always_true_test_is_rejected ... ok
test_bad_ticket_never_reaches_a_model ... ok
test_concerns_escalate_to_next_tier ... ok
test_failing_ticket_is_parked_and_line_keeps_running ... ok
test_removed_ticket_is_skipped ... ok
test_tier_above_ceiling_needs_approval ... ok
test_top_tier_concerns_stay_for_a_human ... ok
----------------------------------------------------------------------
Ran 7 tests in 0.001s
OK
What the real numbers looked like
From my project's log, as of October 30, 2026:
- 1,087 of 1,243 tickets (about 87.4%) passed on the first try.
- 156 (about 12.6%) needed rework. 54 of those actually failed; the rest were escalations, incomplete runs, or blocked.
- The most common reason was a mid-tier model finishing with concerns and the system escalating to a higher tier, about 106 times. Next was failing validation or tests, about 63 times.
If your gate never sends anything back, check the gate.
What this toy leaves out
- Real evidence: here evidence is two booleans. In practice it's test output and logs from the failing run and the passing run, kept with the ticket.
- Real workers: an agent session can drop or time out. In my log, "agent didn't finish" came up about 61 times. A real line needs timeouts and a way to resume.
-
Human confirmation:
DONE_WITH_CONCERNSis not done. Things like load tests that haven't been measured for real still need a person or a real command. - Gate strictness: match it to the work. A refinery producing 99.9999% pure material is a very different plant from one producing 95%. Over-tight gates stop work from coming out at all.
Wrapping up
The model writes the code. The line decides what counts as done. If you're going to let agents run for hours without you, put your effort into the line: small tickets, evidence-based gates, the right model for each job, and a clear place where a human takes over.
Explore the NEXT4I journey and read the original article at: https://go.next4i.com/next4i-devto-en
Originally published at:
https://www.next4i.com/journey/en/next4i-agentic-workflow-02-en
Top comments (0)