work_queue_file.py File Reference

Go to the source code of this file.

Classes
class	oc_orchestrator.work_queue_file.CorruptLogError
class	oc_orchestrator.work_queue_file.QueuePaths
class	oc_orchestrator.work_queue_file.QueueDrift
class	oc_orchestrator.work_queue_file.FileQueueBackend

Namespaces
namespace	oc_orchestrator
namespace	oc_orchestrator.work_queue_file

Functions
str	oc_orchestrator.work_queue_file._now_iso ()
tuple[QueueOp,...]	oc_orchestrator.work_queue_file.read_ops (Path ops_path)
QueueOp	oc_orchestrator.work_queue_file._decode_op (str line, Path path, int number)
str	oc_orchestrator.work_queue_file._encode_op (QueueOp op)
None	oc_orchestrator.work_queue_file.append_op (Path ops_path, QueueOp op)
WorkItem	oc_orchestrator.work_queue_file._item_from_payload (str item_id, Any payload)
str|None	oc_orchestrator.work_queue_file._optional_str (object value)
QueueState	oc_orchestrator.work_queue_file.fold (tuple[QueueOp,...] ops)
str	oc_orchestrator.work_queue_file._annotation (WorkItem item)
list[str]	oc_orchestrator.work_queue_file._bullets (tuple[WorkItem,...] items)
str	oc_orchestrator.work_queue_file._canonical_text (QueueState state)
str	oc_orchestrator.work_queue_file._splice_into_existing (str text, QueueState state)
str	oc_orchestrator.work_queue_file._attribute_for (str title)
str	oc_orchestrator.work_queue_file.render_projection (QueueState state, str|None existing)
QueueState	oc_orchestrator.work_queue_file.parse_projection (str text)
WorkItem	oc_orchestrator.work_queue_file._item_from_annotations (str item_id, str|None annotations)
None	oc_orchestrator.work_queue_file._windows_safe_replace (Path src, Path dst)
tuple[WorkItem, str, int]	oc_orchestrator.work_queue_file._locate (QueueState state, str item_id)

Variables
	oc_orchestrator.work_queue_file.DEFAULT_QUEUE_DIR = Path(".option_c/orchestration")
str	oc_orchestrator.work_queue_file.OPS_FILENAME = "work_queue.ops.jsonl"
str	oc_orchestrator.work_queue_file.PROJECTION_FILENAME = "work_queue.md"
str	oc_orchestrator.work_queue_file.MARKER = "<!-- work-queue: v1 -->"
tuple	oc_orchestrator.work_queue_file.SECTIONS
str	oc_orchestrator.work_queue_file.EMPTY_MARK = "_(none)_"
	oc_orchestrator.work_queue_file._SECTION_RE = re.compile(r"^##\s+(?P<name>Pending|In progress|Completed)\s*$", re.I)
	oc_orchestrator.work_queue_file._BULLET_RE
str	oc_orchestrator.work_queue_file._ACTOR_RECONCILE = "reconcile"