Skip to content
Night Brownie Night Brownie
⌘K

Queue

SQLite-backed task queue for the queue-mediated agent protocol.

All reads and writes use the stdlib sqlite3 module directly — no ORM, no mocks. Tests must use a real temp-file database via pytest tmp_path.

Classes:

  • TaskQueue –

    Durable task queue backed by an SQLite database.

TaskQueue 

TaskQueue(db_path: Path, claim_timeout_seconds: int = 300)

Durable task queue backed by an SQLite database.

Creates the database file and schema on first use. The connection is kept open for the lifetime of the instance.

Parameters:

  • db_path (Path) –

    Filesystem path to the SQLite database file. Intermediate directories are created automatically.

  • claim_timeout_seconds (int, default: 300 ) –

    Seconds before a claimed task without a recent heartbeat is eligible for re-enqueueing.

Methods:

  • __enter__ –

    Return self for use as a context manager.

  • __exit__ –

    Close the connection on context exit.

  • claim_next –

    Claim the oldest pending task for agent_url.

  • close –

    Close the database connection.

  • complete –

    Mark a task completed and store the decision result.

  • drain_completed –

    Return all completed tasks without transitioning their status.

  • enqueue –

    Insert a new task with status=pending.

  • fail_exhausted –

    Mark tasks that have exceeded max_retries as failed.

  • heartbeat –

    Reset last_heartbeat to now, extending the claim window.

  • mark_done –

    Transition a completed task to done after successful processing.

  • requeue_stale –

    Re-enqueue claimed tasks that have exceeded the claim timeout.

__enter__ 

__enter__() -> TaskQueue

Return self for use as a context manager.

__exit__ 

__exit__(*_: object) -> None

Close the connection on context exit.

claim_next 

claim_next(agent_url: str) -> TaskMessage | None

Claim the oldest pending task for agent_url.

Uses a threading lock plus BEGIN IMMEDIATE so that concurrent callers — whether in the same process or different ones — cannot double-claim the same task.

Parameters:

  • agent_url (str) –

    The agent URL requesting a task.

Returns:

  • TaskMessage | None –

    The night_brownie.protocol.TaskMessage for the claimed task,

  • TaskMessage | None –

    or None if the queue has no pending tasks for this agent.

close 

close() -> None

Close the database connection.

complete 

complete(task_id: str, decision: DecisionMessage) -> None

Mark a task completed and store the decision result.

Parameters:

  • task_id (str) –

    ID of the task to mark completed.

  • decision (DecisionMessage) –

    The agent's night_brownie.protocol.DecisionMessage.

drain_completed 

drain_completed() -> list[tuple[TaskMessage, DecisionMessage]]

Return all completed tasks without transitioning their status.

Called by the harness drain loop. The caller must call mark_done for each task after it has been successfully processed, giving at-least-once delivery semantics.

Returns:

enqueue 

enqueue(task: TaskMessage, agent_url: str) -> None

Insert a new task with status=pending.

Parameters:

  • task (TaskMessage) –

    The task message to enqueue.

  • agent_url (str) –

    Base URL of the agent that should process this task.

fail_exhausted 

fail_exhausted(max_retries: int = 3) -> int

Mark tasks that have exceeded max_retries as failed.

Parameters:

  • max_retries (int, default: 3 ) –

    Tasks with retry_count >= max_retries are marked failed.

Returns:

  • int –

    The number of tasks marked failed.

heartbeat 

heartbeat(task_id: str) -> None

Reset last_heartbeat to now, extending the claim window.

Parameters:

  • task_id (str) –

    ID of the claimed task to heartbeat.

mark_done 

mark_done(task_id: str) -> None

Transition a completed task to done after successful processing.

Parameters:

  • task_id (str) –

    ID of the completed task to mark done.

requeue_stale 

requeue_stale() -> int

Re-enqueue claimed tasks that have exceeded the claim timeout.

A task is considered stale when both conditions hold:

  • status = 'claimed'
  • MAX(claimed_at, last_heartbeat) + claim_timeout_seconds < now

Returns:

  • int –

    The number of tasks re-enqueued.