tegwick
88ba666c95
Fresh hub entity per the founder-reviewed decision (not a suggestions rename-bridge): kind: intake per canon/standards/work-record-types_v0.1.md, lifecycle open -> vetted -> routed -> closed(promoted|declined|absorbed). - api/models/base.py::new_uuid7 -- dependency-free RFC 9562 UUIDv7 generator (48-bit ms timestamp, version/variant bits, random remainder); existing tables keep new_uuid (UUIDv4) unchanged, this is opt-in for new work-record entities per the identity-layering canon - api/models/intake.py: Intake + IntakeNote ORM models, mirroring Decision's shape (topic/workplan/repo scope, lane, status, outcome, promoted_to back-link); CHECK constraints enforce scope-required, closed-requires-outcome, promoted-requires-promoted_to at the DB level - migrations/a7c3e9f1b4d2: intakes + intake_notes tables, 3 enum types - api/routers/intake.py: list/create/get/patch + /route + /close + /notes actions, mirroring decisions.py's pattern (409 on invalid transitions, progress event on close) - api/schemas/intake.py: Pydantic create/update/route/close/note schemas - mcp_server/server.py: create_intake, list_intakes, route_intake, close_intake tool wrappers - tests/test_intake.py: 12 tests against the real Postgres test DB (create/list/scope-validation, full lifecycle incl. 409s and the promoted-requires-promoted_to constraint, notes, UUIDv7 verification) Verified live against the running dev API + DB (not just pytest): applied the migration, restarted the MCP server, and ran a full create -> route -> close cycle over the real REST endpoints. No regressions: full existing suite (test_routers_core, test_suggestions, test_mcp_smoke, test_mcp_write_tools, test_mcp_registration, test_consistency_check, test_consistency_sweep) all green. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
53 lines
1.6 KiB
Python
53 lines
1.6 KiB
Python
import os
|
|
import time
|
|
import uuid
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy import DateTime, func
|
|
from sqlalchemy.dialects.postgresql import UUID
|
|
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
|
|
|
|
|
|
class Base(DeclarativeBase):
|
|
pass
|
|
|
|
|
|
class TimestampMixin:
|
|
created_at: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True), server_default=func.now(), nullable=False
|
|
)
|
|
updated_at: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True),
|
|
server_default=func.now(),
|
|
onupdate=func.now(),
|
|
nullable=False,
|
|
)
|
|
|
|
|
|
def new_uuid() -> uuid.UUID:
|
|
return uuid.uuid4()
|
|
|
|
|
|
def new_uuid7() -> uuid.UUID:
|
|
"""Generate a UUIDv7 (RFC 9562): 48-bit big-endian ms timestamp, version
|
|
and variant bits, remaining bits random. Time-sortable, so primary keys
|
|
generated with this helper order chronologically without a separate
|
|
created_at index lookup — the identity layering canon
|
|
(work-record-types_v0.1.md) calls this out as the primary internal key
|
|
for new work-record entities.
|
|
|
|
Dependency-free (no uuid7 in stdlib before Python 3.14, no third-party
|
|
lib added for a ~15-line, non-cryptographic layout).
|
|
"""
|
|
unix_ts_ms = int(time.time() * 1000)
|
|
rand = int.from_bytes(os.urandom(10), "big")
|
|
rand_a = (rand >> 62) & 0x0FFF # top 12 bits of the 80 random bits
|
|
rand_b = rand & 0x3FFFFFFFFFFFFFFF # bottom 62 bits
|
|
value = (
|
|
(unix_ts_ms << 80)
|
|
| (0x7 << 76) # version 7
|
|
| (rand_a << 64)
|
|
| (0x2 << 62) # variant 10
|
|
| rand_b
|
|
)
|
|
return uuid.UUID(int=value)
|