coulomb/state-hub
2
0
Fork 0
generated from coulomb/repo-seed
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Workplan consistency optimization

Browse Source
This commit is contained in:
Bernd Worsch
2026-07-04 00:42:56 +02:00
parent 5388aad77a
commit dbe917ceae
6 changed files with 210 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
api/models/workplan.py
View File

@@ -52,6 +52,12 @@ class Workplan(Base, TimestampMixin):
nullable=True,
index=True,
)
backing_filename: Mapped[str | None] = mapped_column(String(255), nullable=True)
backing_relative_path: Mapped[str | None] = mapped_column(Text, nullable=True)
backing_archived: Mapped[bool | None] = mapped_column(nullable=True)
backing_synced_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True
)
topic: Mapped["Topic | None"] = relationship("Topic", back_populates="workplans") # noqa: F821
repo: Mapped["ManagedRepo"] = relationship("ManagedRepo", lazy="selectin") # noqa: F821

52
api/routers/workstreams.py
View File

@@ -17,6 +17,7 @@ from api.events import EventEnvelope, publish_event
from api.models.managed_repo import ManagedRepo
from api.models.workplan import Workplan
from api.schemas.workplan import (
WorkplanBindingsSync,
WorkplanCreate,
WorkplanRead,
WorkplanUpdate,
@@ -212,9 +213,38 @@ async def _build_workplan_index(session: AsyncSession) -> dict[str, Any]:
"needs_review": bool(review and review.needs_review),
"health_labels": ["needs_review"] if review and review.needs_review else [],
}
await _merge_db_backing_index(session, index)
return {"workplans": index, "workstreams": index}
async def _merge_db_backing_index(session: AsyncSession, index: dict[str, Any]) -> None:
"""Fill index gaps from DB-backed file bindings synced by fix-consistency."""
result = await session.execute(
select(Workplan, ManagedRepo.slug)
.join(ManagedRepo, Workplan.repo_id == ManagedRepo.id)
.where(Workplan.backing_filename.isnot(None))
)
for wp, repo_slug in result.all():
key = str(wp.id)
if key in index:
continue
index[key] = {
"filename": wp.backing_filename,
"relative_path": wp.backing_relative_path,
"repo_slug": repo_slug,
"archived": bool(wp.backing_archived),
"status": normalize_workplan_status(wp.status) if wp.status else None,
"needs_review": False,
"health_labels": [],
}
def _invalidate_workplan_index_cache() -> None:
global _INDEX_CACHE, _INDEX_CACHE_AT
_INDEX_CACHE = None
_INDEX_CACHE_AT = 0.0
def _index_with_meta(*, stale: bool, refresh_in_progress: bool) -> dict[str, Any]:
age = time.monotonic() - _INDEX_CACHE_AT if _INDEX_CACHE_AT else None
return {
@@ -459,6 +489,28 @@ async def workplan_index_preferred(
return await _workplan_index(refresh=refresh, session=session)
@workplan_router.put("/index/bindings")
async def sync_workplan_bindings(
body: WorkplanBindingsSync,
session: AsyncSession = Depends(get_session),
) -> dict[str, int]:
"""Upsert workstation workplan file bindings for remote API index fallback."""
synced_at = datetime.now(timezone.utc)
updated = 0
for entry in body.bindings:
wp = await session.get(Workplan, entry.workplan_id)
if wp is None:
continue
wp.backing_filename = entry.filename
wp.backing_relative_path = entry.relative_path
wp.backing_archived = entry.archived
wp.backing_synced_at = synced_at
updated += 1
await session.commit()
_invalidate_workplan_index_cache()
return {"updated": updated, "received": len(body.bindings)}
@router.post("/", response_model=WorkplanRead, status_code=status.HTTP_201_CREATED)
async def create_workstream(
request: Request,

17
api/schemas/workplan.py
View File

@@ -67,6 +67,19 @@ class WorkplanUpdate(WorkplanStatusMixin):
repo_goal_id: uuid.UUID | None = None
class WorkplanFileBinding(BaseModel):
workplan_id: uuid.UUID
filename: str
relative_path: str
repo_slug: str
archived: bool = False
status: WorkplanStatus | None = None
class WorkplanBindingsSync(BaseModel):
bindings: list[WorkplanFileBinding]
class WorkplanRead(WorkplanStatusMixin):
model_config = ConfigDict(from_attributes=True)
id: uuid.UUID
@@ -87,6 +100,10 @@ class WorkplanRead(WorkplanStatusMixin):
queue_rank: int | None = None
execution_group: str | None = None
scheduled_for: datetime | None = None
backing_filename: str | None = None
backing_relative_path: str | None = None
backing_archived: bool | None = None
backing_synced_at: datetime | None = None
created_at: datetime
updated_at: datetime

31
migrations/versions/f1a2b3c4d5e6_workplan_file_backing.py Normal file
View File

@@ -0,0 +1,31 @@
"""add workplan file backing metadata for remote API index
Revision ID: f1a2b3c4d5e6
Revises: e9f0a1b2c3d4
Create Date: 2026-07-03
"""
from alembic import op
import sqlalchemy as sa
revision = "f1a2b3c4d5e6"
down_revision = "e9f0a1b2c3d4"
branch_labels = None
depends_on = None
def upgrade() -> None:
op.add_column("workplans", sa.Column("backing_filename", sa.String(255), nullable=True))
op.add_column("workplans", sa.Column("backing_relative_path", sa.Text(), nullable=True))
op.add_column("workplans", sa.Column("backing_archived", sa.Boolean(), nullable=True))
op.add_column(
"workplans",
sa.Column("backing_synced_at", sa.DateTime(timezone=True), nullable=True),
)
def downgrade() -> None:
op.drop_column("workplans", "backing_synced_at")
op.drop_column("workplans", "backing_archived")
op.drop_column("workplans", "backing_relative_path")
op.drop_column("workplans", "backing_filename")

51
scripts/consistency_check.py
View File

@@ -570,6 +570,20 @@ def _api_patch(api_base: str, path: str, body: dict) -> Any:
return {"_error": str(exc)}
def _api_put(api_base: str, path: str, body: dict) -> Any:
if not _HAS_HTTPX:
return {"_error": "httpx is not installed"}
if not path.endswith("/"):
path += "/"
try:
with _httpx.Client(base_url=api_base, timeout=30.0, follow_redirects=True) as c:
r = c.put(path, json=body)
r.raise_for_status()
return r.json()
except Exception as exc:
return {"_error": str(exc)}
def _api_post(api_base: str, path: str, body: dict) -> Any:
if not _HAS_HTTPX:
return {"_error": "httpx is not installed"}
@@ -1263,9 +1277,46 @@ def check_repo(api_base: str, repo_slug: str, repo_path_override: str | None = N
# workstream from the file, leaving the first as an invisible orphan.
_check_ghost_duplicates(api_base, workplan_infos, file_ws_ids, report)
_sync_workplan_bindings(api_base, repo_slug, workplan_infos, repo_dir, report)
return report
def _sync_workplan_bindings(
api_base: str,
repo_slug: str,
workplan_infos: list[tuple[Path, dict, str]],
repo_dir: Path,
report: ConsistencyReport,
) -> None:
bindings: list[dict[str, Any]] = []
for wp_file, meta, _ in workplan_infos:
ws_id = str(meta.get("state_hub_workstream_id", "")).strip().strip('"')
if not ws_id:
continue
archived = wp_file.parent.name == "archived"
file_status = normalise_workstream_status(str(meta.get("status", "")).strip())
bindings.append(
{
"workplan_id": ws_id,
"filename": wp_file.name,
"relative_path": workplan_display_path(repo_dir, wp_file),
"repo_slug": repo_slug,
"archived": archived,
"status": file_status or None,
}
)
if not bindings:
return
result = _api_put(api_base, "/workplans/index/bindings", {"bindings": bindings})
if isinstance(result, dict) and "_error" in result:
report.fixes_applied.append(f"bindings WARN: {result['_error']}")
elif isinstance(result, dict):
report.fixes_applied.append(
f"bindings: synced {result.get('updated', 0)}/{result.get('received', len(bindings))}"
)
def _check_orphan_db(
api_base: str,
repo_id: str,

53
tests/test_routers_core.py
View File

@@ -192,6 +192,59 @@ class TestWorkstreams:
assert r.status_code == 200
assert "workstreams" in r.json()
async def test_workplan_bindings_sync_populates_index(self, client, tmp_path):
await _create_domain(client)
topic = await _create_topic(client)
repo = await _create_repo(client, slug="binding-repo", local_path=str(tmp_path))
ws = await _create_workplan(
client,
repo["id"],
topic_id=topic["id"],
slug="binding-wp",
title="Binding WP",
)
workplans_dir = tmp_path / "workplans"
workplans_dir.mkdir()
wp_file = workplans_dir / "BIND-WP-0001-demo.md"
wp_file.write_text(
"---\n"
f"id: BIND-WP-0001\n"
"type: workplan\n"
"title: Binding WP\n"
"status: active\n"
f'state_hub_workstream_id: "{ws["id"]}"\n'
"---\n",
encoding="utf-8",
)
sync = await client.put(
"/workplans/index/bindings",
json={
"bindings": [
{
"workplan_id": ws["id"],
"filename": wp_file.name,
"relative_path": "workplans/BIND-WP-0001-demo.md",
"repo_slug": "binding-repo",
"archived": False,
"status": "active",
}
]
},
)
assert sync.status_code == 200
assert sync.json()["updated"] == 1
hide = await client.patch("/repos/binding-repo", json={"local_path": "/nonexistent/path"})
assert hide.status_code == 200
r = await client.get("/workplans/index?refresh=true")
assert r.status_code == 200
entry = r.json()["workplans"][ws["id"]]
assert entry["filename"] == wp_file.name
assert entry["repo_slug"] == "binding-repo"
# ---------------------------------------------------------------------------
# Task tests