pending_queue

pending_queue

Pending review queue for edits awaiting user approval.

Edits in the review tier (when autonomy mode is tiered) are written here as JSON files. Callers consume the queue via PendingQueue.list() / approve() / reject() to advance edits out of the review tier.

See spec §7.5.

Classes

PendingQueue

PendingQueue(queue_dir: Path)

File-based queue for pending review edits.

Each edit is stored as <queue_dir>/<session_id>__<edit_id>.json.

Source code in src/openjarvis/learning/spec_search/pending_queue.py

def __init__(self, queue_dir: Path) -> None:
    self._dir = Path(queue_dir)
    self._dir.mkdir(parents=True, exist_ok=True)

Functions

enqueue

enqueue(session_id: str, edit: Edit) -> Path

Write an edit to the pending queue. Returns the file path.

Source code in src/openjarvis/learning/spec_search/pending_queue.py

def enqueue(self, session_id: str, edit: Edit) -> Path:
    """Write an edit to the pending queue. Returns the file path."""
    filename = f"{session_id}__{edit.id}.json"
    path = self._dir / filename
    data = {
        "session_id": session_id,
        "edit": json.loads(edit.model_dump_json()),
    }
    path.write_text(json.dumps(data, indent=2), encoding="utf-8")
    logger.info("Enqueued edit %s for review", edit.id)
    return path

list_pending

list_pending() -> list[dict[str, Any]]

Return all pending edits as dicts.

Source code in src/openjarvis/learning/spec_search/pending_queue.py

def list_pending(self) -> list[dict[str, Any]]:
    """Return all pending edits as dicts."""
    results = []
    for path in sorted(self._dir.glob("*.json")):
        try:
            data = json.loads(path.read_text(encoding="utf-8"))
            results.append(data)
        except (json.JSONDecodeError, OSError):
            logger.warning("Skipping corrupt pending file: %s", path)
    return results

get

get(session_id: str, edit_id: str) -> dict[str, Any] | None

Return a specific pending edit, or None.

Source code in src/openjarvis/learning/spec_search/pending_queue.py

def get(self, session_id: str, edit_id: str) -> dict[str, Any] | None:
    """Return a specific pending edit, or None."""
    filename = f"{session_id}__{edit_id}.json"
    path = self._dir / filename
    if not path.exists():
        return None
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except (json.JSONDecodeError, OSError):
        return None

resolve

resolve(session_id: str, edit_id: str) -> bool

Remove a pending edit (approved or rejected). Returns True if found.

Source code in src/openjarvis/learning/spec_search/pending_queue.py

def resolve(self, session_id: str, edit_id: str) -> bool:
    """Remove a pending edit (approved or rejected). Returns True if found."""
    filename = f"{session_id}__{edit_id}.json"
    path = self._dir / filename
    if path.exists():
        path.unlink()
        logger.info("Resolved pending edit %s", edit_id)
        return True
    return False