aaronrene / overseer-kit public
types.py python
266 lines 6.3 KB
Raw
sha256:8d7f41aafae41deee70035a92f93602ef1722a290153597451e5932af503a42c docs: queue board-identity follow-ups so they survive the session Human 12 hours ago
1 """Shared types and exit codes for multi-repo workspace lanes (§MR.7)."""
2
3 from __future__ import annotations
4
5 from dataclasses import dataclass, field
6 from enum import Enum
7 from pathlib import Path
8 from typing import Any, Literal
9
10 EXIT_WORKSPACE_RELAY = 35
11 EXIT_CONFIG = 2
12 EXIT_USAGE = 1
13 EXIT_OK = 0
14
15 WORKSPACE_ROLES = frozenset(
16 {"product_order", "ownership", "enrichment", "edge", "kit", "other"}
17 )
18 SUPPORTED_REGIMES = frozenset({"muse+git-mirror", "muse-only", "git-only"})
19 FORBIDDEN_IDENTITY_KEYS = frozenset(
20 {
21 "x-user-id",
22 "x_user_id",
23 "userid",
24 "user_id",
25 "wallet",
26 "email",
27 "bearer",
28 "token",
29 "password",
30 "secret",
31 "api_key",
32 "apikey",
33 }
34 )
35
36 RelayState = Literal[
37 "not_configured",
38 "ok",
39 "stale_relay",
40 "ambiguous_primary",
41 "missing_primary",
42 "missing_member",
43 "error",
44 ]
45
46 ManifestSource = Literal[
47 "config_manifest",
48 "product_order_root",
49 "local_workspace",
50 "env_override",
51 "home_index",
52 ]
53
54
55 class NextRole(str, Enum):
56 PRIMARY = "primary"
57 RELAY = "relay"
58 PRODUCT_RELAY = "product_relay"
59 LANE_TIP = "lane_tip"
60 ARCHIVED = "archived"
61
62
63 @dataclass(frozen=True)
64 class WorkspaceMemberConfig:
65 """One member row from ``workspace.yaml``."""
66
67 id: str
68 role: str
69 root_raw: str
70 regime: str | None
71 required: bool
72 relay: bool
73 handover: str | None = None
74 roadmap: str | None = None
75 relay_lanes: tuple[str, ...] = ()
76
77
78 @dataclass(frozen=True)
79 class WorkspaceLaneConfig:
80 """One constellation lane."""
81
82 id: str
83 primary: bool
84 owner_member: str | None
85
86
87 @dataclass(frozen=True)
88 class WorkspaceManifest:
89 """Validated constellation manifest (§MR.4.1)."""
90
91 version: int
92 id: str
93 product_order_member: str
94 strict_markers: bool
95 strict_board_names: bool
96 members: tuple[WorkspaceMemberConfig, ...]
97 lanes: tuple[WorkspaceLaneConfig, ...]
98 source_path: Path
99 manifest_source: ManifestSource
100
101 def member(self, member_id: str) -> WorkspaceMemberConfig | None:
102 for row in self.members:
103 if row.id == member_id:
104 return row
105 return None
106
107 def product_order(self) -> WorkspaceMemberConfig:
108 for row in self.members:
109 if row.role == "product_order":
110 return row
111 raise KeyError("product_order member missing")
112
113 def primary_lane(self) -> WorkspaceLaneConfig:
114 for lane in self.lanes:
115 if lane.primary:
116 return lane
117 raise KeyError("primary lane missing")
118
119
120 @dataclass(frozen=True)
121 class NextBlock:
122 """One marked session block extracted from a handover."""
123
124 role: NextRole
125 lane: str | None
126 status: str
127 product_order: str | None
128 tip_hash: str | None
129 heading: str
130 heading_line: int
131 body: str
132 fence: str | None
133 step_id: str | None
134 model: str | None
135 authority: str | None
136 unmarked: bool = False
137
138
139 @dataclass(frozen=True)
140 class MemberBoardPaths:
141 """Resolved handover/roadmap paths for one member."""
142
143 member_id: str
144 root: Path | None
145 present: bool
146 handover_path: Path | None
147 roadmap_path: Path | None
148 handover_basename: str | None
149 roadmap_basename: str | None
150 handover_title: str | None
151 roadmap_title: str | None
152 regime: str | None
153 role: str
154 relay: bool
155 required: bool
156 member_status: Literal["ok", "absent", "missing_required", "error"]
157 board_name_violation: bool = False
158 error: str | None = None
159
160
161 @dataclass(frozen=True)
162 class FreshnessFinding:
163 """One freshness / marker integrity finding."""
164
165 code: RelayState
166 message: str
167 relay_path: str | None = None
168 primary_path: str | None = None
169
170
171 @dataclass(frozen=True)
172 class CheckNextResult:
173 """Outcome of ``ok workspace check-next``."""
174
175 exit_code: int
176 state: RelayState
177 ok: bool
178 lane: str
179 findings: tuple[FreshnessFinding, ...]
180 primary: dict[str, Any] | None = None
181 relays: tuple[dict[str, Any], ...] = ()
182 messages: tuple[str, ...] = ()
183
184
185 @dataclass(frozen=True)
186 class WorkspaceStatusReport:
187 """Structured ``ok workspace status`` payload."""
188
189 configured: bool
190 ok: bool
191 state: RelayState
192 constellation_id: str | None
193 product_order_member: str | None
194 manifest_source: ManifestSource | None
195 manifest_path: str | None
196 authoritative_handover: str | None
197 members: tuple[dict[str, Any], ...]
198 lanes: tuple[dict[str, Any], ...]
199 check_next: dict[str, Any] | None
200 warnings: tuple[str, ...] = ()
201 diagnostics: tuple[str, ...] = ()
202
203 def to_json(self) -> dict[str, Any]:
204 return {
205 "configured": self.configured,
206 "ok": self.ok,
207 "state": self.state,
208 "constellation_id": self.constellation_id,
209 "product_order_member": self.product_order_member,
210 "manifest_source": self.manifest_source,
211 "manifest_path": self.manifest_path,
212 "authoritative_handover": self.authoritative_handover,
213 "members": list(self.members),
214 "lanes": list(self.lanes),
215 "check_next": self.check_next,
216 "warnings": list(self.warnings),
217 "diagnostics": list(self.diagnostics),
218 }
219
220
221 @dataclass(frozen=True)
222 class DoctorFinding:
223 """One doctor diagnostic."""
224
225 code: str
226 message: str
227 member_id: str | None = None
228 path: str | None = None
229
230
231 @dataclass(frozen=True)
232 class DoctorReport:
233 """Outcome of ``ok workspace doctor``."""
234
235 configured: bool
236 findings: tuple[DoctorFinding, ...]
237 ok: bool
238
239 def to_json(self) -> dict[str, Any]:
240 return {
241 "configured": self.configured,
242 "ok": self.ok,
243 "findings": [
244 {
245 "code": f.code,
246 "message": f.message,
247 "member_id": f.member_id,
248 "path": f.path,
249 }
250 for f in self.findings
251 ],
252 }
253
254
255 @dataclass
256 class WorkspaceLoadError(Exception):
257 """Config/manifest load failure (maps to exit 2)."""
258
259 message: str
260 citation: str | None = None
261 exit_code: int = EXIT_CONFIG
262
263 def __str__(self) -> str: # pragma: no cover - trivial
264 if self.citation:
265 return f"{self.message} ({self.citation})"
266 return self.message
File History 1 commit
sha256:8d7f41aafae41deee70035a92f93602ef1722a290153597451e5932af503a42c docs: queue board-identity follow-ups so they survive the session Human 12 hours ago