oc_orchestrator
Loading...
Searching...
No Matches
plan_status.py File Reference

Go to the source code of this file.

Classes

class  oc_orchestrator.plan_status.AgentLookupError
class  oc_orchestrator.plan_status.Status
class  oc_orchestrator.plan_status.Progress
class  oc_orchestrator.plan_status.AcceptanceChecks
class  oc_orchestrator.plan_status.PlanStatus
class  oc_orchestrator.plan_status.WorkerReportError
class  oc_orchestrator.plan_status.EmptySummaryError
class  oc_orchestrator.plan_status.CommsRootNotFoundError

Namespaces

namespace  oc_orchestrator
namespace  oc_orchestrator.plan_status

Functions

str oc_orchestrator.plan_status._default_agent_id_validator (str value)
None oc_orchestrator.plan_status._default_agent_lookup (str|None registry_arg, str agent_id)
datetime oc_orchestrator.plan_status._now_local ()
str oc_orchestrator.plan_status._emit_frontmatter (PlanStatus status)
tuple[PlanStatus, str] oc_orchestrator.plan_status.read (Path path)
None oc_orchestrator.plan_status.write (Path path, PlanStatus status, str body="")
PlanStatus oc_orchestrator.plan_status.update (Path path, str|None body=None, **object fields)
PlanStatus oc_orchestrator.plan_status.transition (Path path, Status new_status, str|None body=None)
None oc_orchestrator.plan_status._validate_transition (Status|None current, Status target)
str oc_orchestrator.plan_status._encode (int value, int length)
str oc_orchestrator.plan_status.new_ulid (datetime now, *, bytes|None entropy=None)
str oc_orchestrator.plan_status.utc_stamp (datetime now)
str oc_orchestrator.plan_status._utc_timestamp (datetime now)
Path oc_orchestrator.plan_status.find_comms_root (Path start)
dict[str, object] oc_orchestrator.plan_status.worker_report_event (*, str plan_id, str agent, str status, Path worktree, Path report_path, str summary, datetime now, bytes|None entropy=None)
Path oc_orchestrator.plan_status.write_worker_report (Path comms_root, dict[str, object] event, *, datetime now)
str oc_orchestrator.plan_status._canonical_agent_id (str value)
Path oc_orchestrator.plan_status._resolve_path (argparse.Namespace args)
Path oc_orchestrator.plan_status._resolve_worktree_field (argparse.Namespace args)
str oc_orchestrator.plan_status._registry_status (Status status)
int oc_orchestrator.plan_status._plan_number (str plan_id)
dict[str, str|None] oc_orchestrator.plan_status._plan_metadata (str plan_id, Path worktree, *, PlanMetadataProvider|None provider=None, PlanPaths|None paths=None)
tuple[StateObserver,...] oc_orchestrator.plan_status._resolve_observers (Sequence[StateObserver]|None observers)
int oc_orchestrator.plan_status._notify (Sequence[StateObserver]|None observers, Callable[[StateObserver], int] emit)
str oc_orchestrator.plan_status._body_note (str|None body, str fallback)
str|None oc_orchestrator.plan_status._status_observed_at (PlanStatus status)
int oc_orchestrator.plan_status._mirror_init (PlanStatus status, *, Sequence[StateObserver]|None observers=None)
int oc_orchestrator.plan_status._mirror_transition (Status previous, PlanStatus current, str|None body, *, Sequence[StateObserver]|None observers=None)
int oc_orchestrator.plan_status._cmd_init (argparse.Namespace args, *, Sequence[StateObserver]|None observers=None, AgentLookup|None lookup=None)
int oc_orchestrator.plan_status._cmd_set (argparse.Namespace args, *, Sequence[StateObserver]|None observers=None)
None oc_orchestrator.plan_status._emit_blocked_event (argparse.Namespace args, PlanStatus current, str|None body)
int oc_orchestrator.plan_status._cmd_show (argparse.Namespace args)
str oc_orchestrator.plan_status._resolve_summary (argparse.Namespace args)
int oc_orchestrator.plan_status._cmd_report (argparse.Namespace args)
int oc_orchestrator.plan_status.main (list[str]|None argv=None, *, str prog=DEFAULT_PROG, Sequence[StateObserver]|None observers=None)

Variables

str oc_orchestrator.plan_status.SCHEMA_VERSION = "2.0"
str oc_orchestrator.plan_status.LEGACY_SCHEMA_VERSION = "1.0"
str oc_orchestrator.plan_status.DEFAULT_FILENAME = ".PLAN_STATUS.md"
str oc_orchestrator.plan_status.WORKSPACE_MARKER = ".option_c_workspace.json"
str oc_orchestrator.plan_status.PROJECT_MARKER = ".option_c"
str oc_orchestrator.plan_status.COMMS_ANCHOR = "oc_chat"
tuple oc_orchestrator.plan_status.COMMS_SUBPATH = (".option_c", "comms")
str oc_orchestrator.plan_status.WORKER_REPORT_LANE = "worker-reports"
str oc_orchestrator.plan_status.WORKER_REPORT_SOURCE = "worker"
str oc_orchestrator.plan_status.WORKER_REPORT_SCHEMA_VERSION = "1.0"
tuple oc_orchestrator.plan_status.WORKER_REPORT_STATUSES = ("awaiting_gate", "blocked")
str oc_orchestrator.plan_status.DEFAULT_REPORT_FILENAME = "report.md"
str oc_orchestrator.plan_status._CROCKFORD = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
 oc_orchestrator.plan_status.AgentLookup = Callable[[str | None, str], None]
 oc_orchestrator.plan_status.AgentIdValidator = Callable[[str], str]
AgentIdValidator oc_orchestrator.plan_status._agent_id_validator = _default_agent_id_validator
 oc_orchestrator.plan_status._split_frontmatter = split_frontmatter
str oc_orchestrator.plan_status.DEFAULT_PROG = "python -m oc_orchestrator.plan_status"