gabriel / musehub public
coord_models.py python
182 lines 8.6 KB
Raw
sha256:8e05daa29ba6702b4a2380a16d690ba31cc099d69c5c859bc6e6f16a0e945f99 Merge 'fix/deploy-memory-limits-and-log-group' into 'dev' —… Human 16 days ago
1 """SQLAlchemy ORM model for the MuseHub coordination bus.
2
3 The coordination bus stores advisory coordination records pushed by Muse CLI
4 agents (reservations, intents, releases, heartbeats, dependencies) so that
5 agent swarms running on different machines can share state without filesystem
6 access to each other's ``.muse/coordination/`` directories.
7
8 Design
9 ------
10 - **Write-once with one exception**: all record kinds are written once and
11 never updated, *except* ``heartbeat`` records which are upserted on each
12 push (the ``run_id`` keeps them distinct per agent).
13 - **Auto-increment cursor**: the ``id`` column is a monotonically increasing
14 integer primary key. SSE watch clients use ``?since_id=<id>`` to resume
15 polling after the last event they received — more reliable than timestamps
16 because writes within the same millisecond are still totally ordered.
17 - **Repo-scoped**: all records are scoped to a single ``repo_id`` (FK →
18 ``musehub_repos.repo_id``). Different repos have independent namespaces.
19 - **Kind + ID uniqueness**: ``(repo_id, kind, record_id)`` is unique —
20 the same coordination record cannot be pushed twice (idempotent re-push
21 returns ``skipped`` rather than a 409 error).
22 - **Payload JSON**: arbitrary JSON blob carrying the full coordination record
23 as serialized by the Muse CLI (the same format stored locally in ``.muse/``).
24
25 Indexes
26 -------
27 - ``ix_coord_repo_id_cursor``: ``(repo_id, id)`` — primary pull query pattern:
28 "give me all records for repo X with id > Y in insertion order".
29 - ``ix_coord_repo_kind_id``: ``(repo_id, kind, id)`` — filtered pull query:
30 "give me all heartbeats for repo X with id > Y".
31 """
32
33 from datetime import datetime, timezone
34
35 from sqlalchemy import ARRAY, DateTime, ForeignKey, Index, Integer, PrimaryKeyConstraint, String, UniqueConstraint
36 from sqlalchemy.orm import Mapped, mapped_column
37 from sqlalchemy.dialects.postgresql import JSONB
38
39 from musehub.types.json_types import JSONObject, JSONValue # JSONValue needed for ForwardRef resolution in Mapped[]
40 from musehub.db.database import Base
41
42 def _utc_now() -> datetime:
43 return datetime.now(tz=timezone.utc)
44
45 class MusehubCoordRecord(Base):
46 """A single coordination record pushed from a Muse CLI agent.
47
48 One row per (repo_id, kind, record_id) triple. Re-pushing the same
49 triple is silently skipped (idempotent). ``heartbeat`` records are
50 upserted rather than skipped — the same agent re-pushes its heartbeat
51 repeatedly, updating ``payload`` and ``created_at`` in place.
52
53 Attributes:
54 id: Auto-increment integer PK — used as the SSE cursor.
55 repo_id: FK → ``musehub_repos.repo_id``.
56 kind: Coordination record type — one of:
57 ``reservation``, ``intent``, ``release``, ``heartbeat``,
58 ``dependency``, ``task``, ``claim``.
59 record_id: ID of the original local coordination record — a sha256
60 genesis ID (``sha256:<64-hex>`` = 71 chars) or an opaque
61 run_id string. Maximum 128 chars.
62 run_id: Agent/pipeline identifier (opaque string, max 255 chars).
63 payload: Full coordination record as a JSON object.
64 created_at: Server-side write timestamp (UTC).
65 expires_at: Optional expiry timestamp from the original record (UTC).
66 ``NULL`` for records that never expire (dependencies,
67 tasks, claims).
68 """
69
70 __tablename__ = "musehub_coord_records"
71 __table_args__ = (
72 UniqueConstraint(
73 "repo_id", "kind", "record_id",
74 name="uq_coord_repo_kind_uuid",
75 ),
76 Index("ix_coord_repo_id_cursor", "repo_id", "id"),
77 Index("ix_coord_repo_kind_id", "repo_id", "kind", "id"),
78 )
79
80 id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
81 repo_id: Mapped[str] = mapped_column(
82 String(128),
83 ForeignKey("musehub_repos.repo_id", ondelete="CASCADE"),
84 nullable=False,
85 index=True,
86 )
87 kind: Mapped[str] = mapped_column(String(32), nullable=False)
88 record_id: Mapped[str] = mapped_column(String(128), nullable=False)
89 run_id: Mapped[str] = mapped_column(String(255), nullable=False, default="")
90 payload: Mapped[JSONObject] = mapped_column(JSONB, nullable=False)
91 created_at: Mapped[datetime] = mapped_column(
92 DateTime(timezone=True), nullable=False, default=_utc_now
93 )
94 expires_at: Mapped[datetime | None] = mapped_column(
95 DateTime(timezone=True), nullable=True, default=None
96 )
97
98 def to_dict(self) -> JSONObject:
99 """Serialize to a JSON-safe dict for API responses."""
100 return {
101 "id": self.id,
102 "repo_id": self.repo_id,
103 "kind": self.kind,
104 "record_id": self.record_id,
105 "run_id": self.run_id,
106 "payload": self.payload,
107 "created_at": self.created_at.isoformat() if self.created_at else None,
108 "expires_at": self.expires_at.isoformat() if self.expires_at else None,
109 }
110
111 class MusehubCoordReservation(Base):
112 """Persistent store for muse coord reserve events synced from agents.
113
114 Advisory symbol locks. Agents reserve symbols before editing to signal
115 intent and enable conflict forecasting on the MuseHub coordination dashboard.
116 Released explicitly or when TTL expires.
117
118 Distinct from ``MusehubCoordRecord`` (raw event bus) — this table is the
119 *materialized* view of active reservations, queryable by symbol address.
120 """
121
122 __tablename__ = "musehub_coord_reservations"
123 __table_args__ = (
124 PrimaryKeyConstraint("reservation_id", "symbol_address"),
125 Index("ix_coord_reservations_repo_id", "repo_id"),
126 )
127
128 # Composite PK: one row per (reservation_id, symbol_address) — a single reservation
129 # covers N symbol addresses, producing N rows that share the same reservation_id.
130 reservation_id: Mapped[str] = mapped_column(String(128), nullable=False)
131 symbol_address: Mapped[str] = mapped_column(String(512), nullable=False)
132 repo_id: Mapped[str] = mapped_column(
133 String(128),
134 ForeignKey("musehub_repos.repo_id", ondelete="CASCADE"),
135 nullable=False,
136 )
137 agent_id: Mapped[str] = mapped_column(String(255), nullable=False)
138 agent_model_id: Mapped[str] = mapped_column(String(255), nullable=False, default="")
139 ttl_s: Mapped[int] = mapped_column(Integer, nullable=False, default=300)
140 created_at: Mapped[datetime] = mapped_column(
141 DateTime(timezone=True), nullable=False, default=_utc_now
142 )
143 expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
144 released_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
145
146 class MusehubCoordTask(Base):
147 """Persistent task queue mirroring muse coord enqueue/claim/complete.
148
149 Enables agentception and other orchestrators to dispatch work to agents
150 through MuseHub with full visibility into queue state and task outcomes.
151
152 ``status`` lifecycle: pending → claimed → completed | failed
153 ``depends_on`` is a JSON list of task_id strings that must complete first.
154 ``run_id`` is an agentception run reference (opaque string).
155 """
156
157 __tablename__ = "musehub_coord_tasks"
158 __table_args__ = (
159 Index("ix_coord_tasks_repo_queue_status", "repo_id", "queue", "status"),
160 Index("ix_coord_tasks_repo_priority", "repo_id", "priority"),
161 )
162
163 task_id: Mapped[str] = mapped_column(String(128), primary_key=True)
164 repo_id: Mapped[str] = mapped_column(
165 String(128),
166 ForeignKey("musehub_repos.repo_id", ondelete="CASCADE"),
167 nullable=False,
168 index=True,
169 )
170 queue: Mapped[str] = mapped_column(String(255), nullable=False, default="default", server_default="default")
171 priority: Mapped[int] = mapped_column(Integer, nullable=False, default=50)
172 payload: Mapped[JSONObject] = mapped_column(JSONB, nullable=False)
173 # pending | claimed | completed | failed
174 status: Mapped[str] = mapped_column(String(20), nullable=False, default="pending", server_default="pending", index=True)
175 claimed_by: Mapped[str | None] = mapped_column(String(255), nullable=True)
176 claimed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
177 completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
178 depends_on: Mapped[list[str]] = mapped_column(ARRAY(String(128)), nullable=False, default=list)
179 run_id: Mapped[str | None] = mapped_column(String(255), nullable=True)
180 created_at: Mapped[datetime] = mapped_column(
181 DateTime(timezone=True), nullable=False, default=_utc_now
182 )
File History 3 commits
sha256:8e05daa29ba6702b4a2380a16d690ba31cc099d69c5c859bc6e6f16a0e945f99 Merge 'fix/deploy-memory-limits-and-log-group' into 'dev' —… Human 16 days ago
sha256:3fadb0439bba9451b89229676971c0d4a40900dec7810e9d5f8791b8d950d505 fix: install.sh version from latest published tarball, not … Sonnet 4.6 minor ⚠ 107 days ago
sha256:763eb2cb8675073b84c19345b27586d2ed939a9aee97c5479b69f502f1a70eff fix(tests): update test suite to match current implementation Sonnet 4.6 patch 128 days ago