agent_task_queues v1.0.0

name: agent_task_queues description: "Run a durable task backlog for agent work: keep one queue file with a fixed record shape, apply intake and priority rules, hold work in progress limits, enforce claim and release discipline, handle aging and stale tasks, and close work only against written done criteria. Use when agent work needs a standing backlog that survives a single run, when tasks pile up without owners, or when in progress work stalls and ages without review." metadata: { "includeInPrompt": true }

Agent Task Queues

Purpose

Give agent work a standing backlog that stays readable over time. The backlog lives in one durable queue file. Each task carries the same record shape, enters through the same intake rules, moves in priority order, and closes only when its written done criteria are met. Claim and release rules keep each task with one claimant at a time, work in progress limits keep the active set small, and a regular aging pass keeps old tasks from sitting unseen.

Scope:

• message-queues covers messaging infrastructure (when to queue, dead letters, brokers). That layer moves messages between systems.
• subagent-orchestration and multi-agent-orchestration cover fanning work out to workers during a run and reconciling results. That layer runs work inside a single run.
• This skill is the standing work backlog itself: intake shape, priority rules, work-in-progress limits, aging and stale-task handling, claim and release discipline, done criteria. Keep the layer clear: backlog management, not message transport and not run-time fan-out.
• This skill is file based. It works on a queue file in JSON form, with plain file and shell operations only. It names no queue product, no broker, and no platform API.
• Verification note, 2026-10-21: queue_check.py below was run with Python 3.12.3, the version python3 --version returned on the drafting machine, against a fixture built by the script itself with --make-fixture fixture, followed by python3 queue_check.py fixture/queue.json. The fixture run returned tasks_total: 9, count_queued: 3, count_in_progress: 4, count_blocked: 1, count_done: 1, count_cancelled: 0, wip_over_by: 1, stale_total: 3, missing_done_criteria_total: 2, duplicate_ids_total: 1, verdict: REVIEW. The script uses only the Python standard library (json, os, and sys, plus collections for counting), so no package version is quoted and no install step is needed. What was NOT verified: no specific queue product, broker, or platform API is claimed, no product behavior was tested, and no claim is made about how any product stores or orders its tasks. The skill is file based, and only the file based checker and the counts above were verified by a real run.

Workflow

1. Keep one queue file as the backlog of record.

• Hold all open and closed agent work in one queue file, in JSON form: either a list of task records, or an object with a tasks list. Splitting the backlog across scraps of notes, chat threads, and run logs makes counts, ages, and ownership impossible to check.
• Write changes back to the same file at the end of each work pass. A change that lives only in a run summary has not landed in the backlog.
• Keep closed tasks in the file. Done and cancelled records are the history that later counts and duplicate checks rely on. Prune them only on a stated schedule, and record the prune in the file header notes or a dated line in the review record.
• Give the file a plain location and a plain name, such as queue.json, and use that one file in every step below and in the checker run.

1. Give every task the same record shape.

• id: a short unique identifier, such as T-001. Ids are assigned at intake, never reused, and never changed after work starts.
• title: one plain line that names the work. A title that needs a paragraph to explain belongs in the task notes, not in the title field.
• state: one of queued, in_progress, blocked, done, cancelled. No other values. A state outside that set signals a record error, not a new stage.
• priority: one of P0, P1, P2, P3. P0 is the most urgent. Every task carries a priority from intake onward.
• age_days: whole days since intake, refreshed at each review pass before the checker runs. Age is measured from intake, not from the last edit, so a task that keeps getting touched still shows its true age.
• owner: the current claimant, or an empty string when the task is unclaimed. The field holds one value only. Shared work is split into separate tasks, each with its own claimant.
• done_criteria: a written statement of what counts as done for this task, in checkable form. Step 9 covers the test for good criteria.
• A record missing a field is incomplete. Fix the record at intake or at the review pass that finds the gap. Leave no field to be guessed from the title.

1. Apply intake rules before a task enters the queue.

• Intake takes a task only when the record is complete: id, title, state set to queued, priority, age_days set to 0, owner empty or named, and done_criteria written. A request with no done criteria goes back for criteria before it enters.
• Check the id against the file before adding. A duplicate id is a stop, not a merge: settle which record keeps the id, give the other record a new id, and note the change in the review record.
• Check for duplicate work, which is a separate problem from a duplicate id. Read the titles of queued and in_progress tasks. When the same work already exists, add the new detail to the existing record and close the new intake as a note on that record instead of a second task.
• Break large requests at intake. A task that names several separate outcomes becomes several tasks, each with its own done criteria, so each part can be claimed, aged, and closed on its own.
• Cap intake claims. A priority of P0 at intake needs a stated reason in the task record. Without a reason, intake sets the next lower priority and leaves the ordering in step 4 to sort it out.

1. Order the queue by priority, then by age.

• Claim order is fixed: P0 first, then P1, then P2, then P3. Within one priority, the oldest task goes first. Age breaks ties so quiet tasks rise instead of sinking under newer work of the same weight.
• Priority changes are recorded changes. When a task changes priority, write the new value, the date, and the reason in the review record. Silent reordering hides the decision that caused it.
• Blocked tasks keep their priority. A blocked P0 still outranks a queued P1 when the blocker clears, so the order survives the block.
• Starvation guard: at each review pass, look at the oldest queued task in each priority band. A low priority task that has aged past the stale threshold in step 8 gets a direct decision at that pass: raise its priority, schedule it, or cancel it with a reason. Leaving it in place with no decision counts as a stale handling miss.

1. Set and hold work in progress limits.

• The work in progress limit caps the count of tasks in the in_progress state at any one time. The checker default is 3. Set the limit in writing before work starts, sized to the claimants actually available, and change it only as a recorded decision.
• Blocked tasks do not count toward the limit. A blocked task has stopped, and counting it as active work would hide the room that is really free.
• When the limit is reached, claiming stops. The next task moves only when an in_progress task reaches done, moves to blocked with a named blocker, or is released back to queued under step 7.
• The checker reports wip_in_progress and wip_over_by for each run. A breach is a finding to act on in the same pass: finish, block, or release tasks until the count is at or under the limit. A breach carried across two review passes without action is a backlog fault, not a busy period.
• Watch the per claimant load as a reading, even though the checker counts the total. One claimant holding most of the active set is a stall risk, because one stall then stops most of the queue.

1. Enforce claim discipline: one claimant at a time.

• A task is claimed by writing a claimant value into owner and moving state from queued to in_progress, in the same edit. A claim that changes only one of the two fields is half done and leaves the record unclear.
• Only unclaimed queued tasks can be claimed, and only in the order from step 4, unless a priority change has been recorded. Skipping the order to take easier work first is a discipline fault, and the skipped tasks keep aging while it happens.
• A claimed task has one claimant. Handing work over means the current claimant releases under step 7 first, and the next claimant claims after. Two names in the owner field, or work proceeding while the field names someone else, both break the record.
• A claim is a commitment to start now. Claiming a task to reserve it, with no start in the same work pass, holds a slot in the work in progress count and blocks the order for other claimants. Claim at the point of starting.
• Keep the claim visible. The queue file is the claim record. A claim made in chat or in a run log that never reaches the file does not exist for the next pass.

1. Enforce release discipline: release on stall, block, or handoff.

• Release means clearing owner and moving state back to queued, with a short note in the task record stating why and what state the work is in. The next claimant reads that note before claiming.
• Release when work stalls: no progress across a full work pass, no clear next step, or the claimant is pulled onto other work. A stalled task held in_progress counts against the limit and ages in place. Releasing it costs little. Holding it costs the slot.
• A task that is waiting on a named outside condition is blocked, not stalled. Move it to blocked, keep the claimant in owner, and write the blocker and its date in the task record. Blocked is a stated condition with a name. Stalled is no movement and no stated condition.
• Review blocked tasks at every review pass. When the blocker clears, the task returns to in_progress if the limit has room, or to queued at its priority if the limit is full.
• On handoff, the releasing claimant writes the note before the release, not after. State what is done, what is left, and where any partial output lives, in plain terms the next claimant can act on without asking.

1. Run an aging pass and handle stale tasks directly.

• Refresh age_days for every open task at the start of the pass, counting from the intake date in the record. Then run the checker to get the stale list.
• A task is stale when it sits in an open state (queued, in_progress, or blocked) with age_days past the stale threshold. The checker default is 14 days. Done and cancelled tasks are closed and never count as stale, however old they are.
• Each stale task gets one of four handlings, recorded in the review record: continue (work resumes now, with a next step named), requeue (released under step 7 so a fresh claimant can take it), re-scope (title, done criteria, or priority rewritten to match the work that is still worth doing), or cancel (closed under step 9 with a reason).
• A stale blocked task needs its blocker rechecked first. If the blocker has cleared, the task is not stale work, it is waiting work that missed its restart. Restart it or release it at the same pass.
• Report the aging result in counts: stale total at the start of the pass, and how many took each handling. A stale total that never falls across passes means the pass is listing tasks without deciding them.

1. Close work only against its done criteria, with a closure check.

• Good done criteria are checkable: a reader who did not do the work can test each one against a result and say yes or no. Criteria written as effort, such as work on the report, fail this test and get rewritten at intake or at the review pass that spots them.
• At closure, read the done criteria line by line against what was produced. Each line is met or unmet. The state moves to done only when every line is met.
• When part of the criteria is met and part is not, the task stays open. Split the unmet part into a new task with its own id and criteria, or keep the original task open with a note on what remains. Marking a partly met task done hides the remainder from counts and ordering.
• Cancellation is a closure too, and it needs a reason in the task record: duplicate of another task (named by id), no longer needed, or superseded by another task (named by id). A cancelled task keeps its record and its age. Deleting the record erases the history.
• After closure, the claimant field stays as the record of who closed it until the next prune. The closed task leaves the work in progress count at once, which is what frees the slot for the next claim in step 5.

1. Run the checker and review the backlog on a fixed cadence.

• Save the script from the Operating Rules section as queue_check.py. At each review pass, run it against the queue file: python3 queue_check.py queue.json, adding --wip-limit N or --stale-days N when the recorded settings differ from the defaults of 3 and 14.
• Work the checker output in a fixed order: duplicate ids first, since counts cannot be trusted while one id names two records; then missing done criteria, since open tasks without criteria cannot close cleanly; then the work in progress breach; then the stale list from step 8.
• Keep the full checker output with the review record. Counts quoted later come from that output, not from memory of the file.
• Set the cadence in writing: a review pass on a fixed interval, plus a pass whenever intake adds a batch of tasks or a breach appears. The cadence is short enough that a task cannot pass the stale threshold between passes without being seen.
• End the pass with the deliverables in the Output Contract, in order. A pass that fixes the file but leaves no record has no proof the checks ran.

Output Contract

Deliver these items, in this order:

1. Updated queue file: the queue file after the pass, with every record in the shape from Workflow step 2, claims and releases from steps 6 and 7 written into owner and state, and closures from step 9 recorded as done or cancelled with reasons.
2. Checker output: the full output of queue_check.py for the run. On the worked fixture in the Operating Rules section the headline numbers are: tasks_total: 9, count_queued: 3, count_in_progress: 4, count_blocked: 1, count_done: 1, count_cancelled: 0, wip_over_by: 1, stale_total: 3, missing_done_criteria_total: 2, duplicate_ids_total: 1, verdict: REVIEW.
3. Priority order list: the claimable queued tasks in claim order from Workflow step 4, priority first and age second, with the work in progress room stated at the top (limit, current in_progress count, slots free).
4. Claim and release log: one row per claim, release, block, and handoff in the pass, with the task id, the change made, the claimant before and after, and the reason for any release or block.
5. Aging report: the stale list from the checker, with the handling each stale task took under Workflow step 8 (continue, requeue, re-scope, or cancel), the stale total at the start of the pass, and the count per handling.
6. Closure list: one row per task closed in the pass, with the task id, the done criteria checked line by line with met or unmet per line, and for cancellations the reason and any task named as the duplicate or replacement.
7. Limits: anything the pass did not cover, stated plainly (for example records left incomplete, a stale task left without a handling, or a threshold setting that was assumed rather than recorded).

Operating Rules

1. One file is the backlog. Work that is not in the queue file is not in the backlog, and counts, ages, and claims are read from the file only.
2. One task, one claimant, one state. The owner field holds a single claimant or sits empty, the state holds one value from the fixed set, and a claim or release changes owner and state together.
3. No task enters or stays in progress without written done criteria. A task found without criteria gets criteria written at that pass, or it returns to intake.
4. Priority order governs claiming. Priority first, age second. Changes to priority or order are recorded with a reason, never applied silently.
5. The work in progress limit is a hard cap. At the limit, claiming stops until a task closes, blocks with a named blocker, or releases. A breach is cleared in the pass that finds it.
6. Release on stall is normal work, not a failure mark. A released task keeps its note on what state the work is in, and that note is written before the release.
7. Stale tasks get a decision at the pass that lists them: continue, requeue, re-scope, or cancel, each recorded. Listing a stale task again at the next pass with no decision is the fault the aging pass exists to prevent.
8. Closure reads the criteria, not the effort spent. Done means every done criteria line tested met. Partly met work stays open or splits. Cancelled work keeps its record and its reason.
9. Every number in the deliverable comes from the checker output or a direct read of the queue file in the same pass. Numbers from memory of an earlier pass are stale until the checker produces them again.
10. Use the checker below for the mechanical checks. Save it as queue_check.py, run it as its header describes, and put its full output in the deliverable as item 2 of the Output Contract. Report numbers from its output, not from memory of the file.

Queue check (queue_check.py)

Run: python3 queue_check.py --make-fixture fixture builds the worked fixture in a folder named fixture, holding a queue.json file. Then run: python3 queue_check.py fixture/queue.json to reproduce the output below. Run it on a real backlog with python3 queue_check.py QUEUE_FILE and add --wip-limit N or --stale-days N when the recorded settings differ from the defaults in the script header.

Expected output for the fixture run:

text
queue_file: fixture/queue.json
tasks_total: 9
count_queued: 3
count_in_progress: 4
count_blocked: 1
count_done: 1
count_cancelled: 0
wip_limit: 3
wip_in_progress: 4
wip_over_by: 1
stale_threshold_days: 14
stale_total: 3
STALE id=T-003 state=in_progress age_days=18
STALE id=T-006 state=blocked age_days=21
STALE id=T-007 state=queued age_days=16
missing_done_criteria_total: 2
MISSING_DONE id=T-004
MISSING_DONE id=T-007
duplicate_ids_total: 1
DUPLICATE id=T-002 count=2
verdict: REVIEW

The fixture is built to show each checker result in a small set: nine tasks in total, with three queued, four in progress, one blocked, and one done. The four in progress tasks sit one over the limit of three, which gives wip_over_by: 1. Three open tasks sit past the stale threshold of 14 days, at 16, 18, and 21 days. Two tasks carry empty done criteria, and one id, T-002, appears on two records, which gives duplicate_ids_total: 1 and a verdict of REVIEW.

queue_check.py (uses only the Python standard library). Save this code as queue_check.py:

python
#!/usr/bin/env python3
"""Validate an agent task queue file and report backlog health.

Usage:
  python3 queue_check.py --make-fixture DIR
  python3 queue_check.py QUEUE_FILE [--wip-limit N] [--stale-days N]

The queue file is JSON: either a list of task records, or an
object with a "tasks" list. Each task record uses these fields:
  id             string, unique task identifier
  title          string
  state          one of: queued, in_progress, blocked, done, cancelled
  priority       one of: P0, P1, P2, P3
  age_days       integer, days since intake
  owner          string, the current claimant, or empty when unclaimed
  done_criteria  string, what counts as done for this task

Defaults: --wip-limit 3 caps the in_progress count,
--stale-days 14 marks open tasks older than the threshold as
stale. Open states are queued, in_progress, and blocked. Done
and cancelled tasks are closed and never count as stale.

Output prints counts by state, the WIP check, stale tasks,
tasks missing done criteria, duplicate ids, and a verdict.
The checker reads only. It changes no queue file.

Exit code is 0 when the queue file was read and checked, 1
when it cannot be read or parsed. Standard library only.
"""
import json
import os
import sys
from collections import Counter

STATES = ["queued", "in_progress", "blocked", "done", "cancelled"]
OPEN_STATES = {"queued", "in_progress", "blocked"}
DEFAULT_WIP_LIMIT = 3
DEFAULT_STALE_DAYS = 14


def check(queue_path, wip_limit, stale_days):
    if not os.path.isfile(queue_path):
        print(f"error: queue file not found: {queue_path}")
        return 1
    try:
        with open(queue_path, "r", encoding="utf-8") as fh:
            data = json.load(fh)
    except (OSError, json.JSONDecodeError) as exc:
        print(f"error: cannot read queue file: {exc}")
        return 1
    if isinstance(data, dict):
        tasks = data.get("tasks", [])
    else:
        tasks = data
    if not isinstance(tasks, list):
        print("error: queue data is not a list of tasks")
        return 1
    counts = Counter()
    for task in tasks:
        state = str(task.get("state", "")).strip()
        counts[state if state in STATES else "unknown"] += 1
    id_counts = Counter(str(task.get("id", "")).strip() for task in tasks)
    duplicates = sorted(k for k, v in id_counts.items() if k and v > 1)
    stale = []
    missing_done = []
    for task in tasks:
        tid = str(task.get("id", "")).strip()
        state = str(task.get("state", "")).strip()
        try:
            age = int(task.get("age_days", 0))
        except (TypeError, ValueError):
            age = 0
        if state in OPEN_STATES and age > stale_days:
            stale.append((tid, state, age))
        criteria = str(task.get("done_criteria", "") or "").strip()
        if state != "cancelled" and not criteria:
            missing_done.append(tid)
    stale.sort(key=lambda r: r[0])
    missing_done.sort()
    wip = counts.get("in_progress", 0)
    wip_over = max(0, wip - wip_limit)
    print(f"queue_file: {queue_path}")
    print(f"tasks_total: {len(tasks)}")
    for state in STATES:
        print(f"count_{state}: {counts.get(state, 0)}")
    if counts.get("unknown", 0):
        print(f"count_unknown: {counts.get('unknown', 0)}")
    print(f"wip_limit: {wip_limit}")
    print(f"wip_in_progress: {wip}")
    print(f"wip_over_by: {wip_over}")
    print(f"stale_threshold_days: {stale_days}")
    print(f"stale_total: {len(stale)}")
    for tid, state, age in stale:
        print(f"STALE id={tid} state={state} age_days={age}")
    print(f"missing_done_criteria_total: {len(missing_done)}")
    for tid in missing_done:
        print(f"MISSING_DONE id={tid}")
    print(f"duplicate_ids_total: {len(duplicates)}")
    for tid in duplicates:
        print(f"DUPLICATE id={tid} count={id_counts[tid]}")
    problems = wip_over + len(stale) + len(missing_done) + len(duplicates)
    print(f"verdict: {'PASS' if problems == 0 else 'REVIEW'}")
    return 0


def make_fixture(base):
    os.makedirs(base, exist_ok=True)
    tasks = [
        {"id": "T-001", "title": "Audit intake form", "state": "queued",
         "priority": "P1", "age_days": 3, "owner": "",
         "done_criteria": "Intake fields match the record shape"},
        {"id": "T-002", "title": "Draft priority rules", "state": "in_progress",
         "priority": "P0", "age_days": 5, "owner": "agent-a",
         "done_criteria": "Rules listed and checked against the queue"},
        {"id": "T-003", "title": "Fix claim release step", "state": "in_progress",
         "priority": "P1", "age_days": 18, "owner": "agent-b",
         "done_criteria": "Release path tested on a stalled task"},
        {"id": "T-004", "title": "Review aging report", "state": "in_progress",
         "priority": "P2", "age_days": 2, "owner": "agent-a",
         "done_criteria": ""},
        {"id": "T-005", "title": "Close finished migration", "state": "in_progress",
         "priority": "P1", "age_days": 9, "owner": "agent-c",
         "done_criteria": "Closure checks pass and state is done"},
        {"id": "T-006", "title": "Triage blocked item", "state": "blocked",
         "priority": "P0", "age_days": 21, "owner": "agent-b",
         "done_criteria": "Blocker named, dated, and routed"},
        {"id": "T-007", "title": "Write closure checklist", "state": "queued",
         "priority": "P2", "age_days": 16, "owner": "",
         "done_criteria": ""},
        {"id": "T-008", "title": "Ship queue report", "state": "done",
         "priority": "P1", "age_days": 30, "owner": "agent-a",
         "done_criteria": "Report posted to the queue record"},
        {"id": "T-002", "title": "Draft priority rules copy", "state": "queued",
         "priority": "P3", "age_days": 1, "owner": "",
         "done_criteria": "Rules listed and checked against the queue"},
    ]
    queue_path = os.path.join(base, "queue.json")
    with open(queue_path, "w", encoding="utf-8") as fh:
        json.dump({"tasks": tasks}, fh, indent=2)
        fh.write("\n")
    print(f"fixture_queue: {queue_path}")
    print(f"fixture_tasks: {len(tasks)}")
    return 0


def main():
    args = sys.argv[1:]
    if len(args) == 2 and args[0] == "--make-fixture":
        return make_fixture(args[1])
    wip_limit = DEFAULT_WIP_LIMIT
    stale_days = DEFAULT_STALE_DAYS
    positional = []
    i = 0
    while i < len(args):
        if args[i] == "--wip-limit" and i + 1 < len(args):
            wip_limit = int(args[i + 1])
            i += 2
        elif args[i] == "--stale-days" and i + 1 < len(args):
            stale_days = int(args[i + 1])
            i += 2
        else:
            positional.append(args[i])
            i += 1
    if len(positional) != 1:
        print("usage: queue_check.py QUEUE_FILE [--wip-limit N] [--stale-days N]")
        print("       queue_check.py --make-fixture DIR")
        return 2
    return check(positional[0], wip_limit, stale_days)


if __name__ == "__main__":
    raise SystemExit(main())