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

Record deferred reconciliation requests

Browse Source
This commit is contained in:
Bernd Worsch
2026-05-23 17:50:27 +02:00
parent 141e9cca1d
commit 997766e99d
4 changed files with 134 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
api/routers/reconciliation.py
View File

@@ -5,6 +5,7 @@ from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from api.database import get_session
from api.models.agent_message import AgentMessage
from api.models.managed_repo import ManagedRepo
from api.models.task import Task
from api.models.task import TaskStatus
@@ -37,6 +38,39 @@ async def _workstream_tasks_terminal(session: AsyncSession, workstream_id: uuid.
return bool(statuses) and all(status in {"done", "cancelled"} for status in statuses)
def _deferred_message(
*,
body: StateChangeRequest,
current_status: str,
target_status: str,
classification: ReconciliationClass,
reason: str,
follow_up: str,
workplan_path: str | None,
) -> AgentMessage:
subject = f"Reconcile {body.target_type} state change: {current_status} -> {target_status}"
lines = [
"A UI-originated state change could not be written through immediately.",
"",
f"target_type: {body.target_type}",
f"target_id: {body.target_id}",
f"actor: {body.actor}",
f"intent: {body.intent or ''}",
f"current_status: {current_status}",
f"target_status: {target_status}",
f"reconciliation_class: {classification.value}",
f"reason: {reason}",
f"follow_up: {follow_up}",
f"workplan_path: {workplan_path or ''}",
]
return AgentMessage(
from_agent=body.actor,
to_agent="state-hub",
subject=subject,
body="\n".join(lines),
)
@router.post("/state-change", response_model=StateChangeResponse)
async def classify_state_change(
body: StateChangeRequest,
@@ -76,6 +110,7 @@ async def classify_state_change(
tasks_terminal=tasks_terminal,
)
write_result = "not_attempted"
reconciliation_record_id = None
if body.apply:
if classification.reconciliation_class == ReconciliationClass.WRITE_THROUGH and workplan_ref:
patch_workplan_status(workplan_ref.path, target_status)
@@ -83,6 +118,19 @@ async def classify_state_change(
await session.commit()
write_result = "applied"
else:
msg = _deferred_message(
body=body,
current_status=current_status,
target_status=target_status,
classification=classification.reconciliation_class,
reason=classification.reason,
follow_up=classification.follow_up,
workplan_path=workplan_ref.relative_path if workplan_ref else None,
)
session.add(msg)
await session.commit()
await session.refresh(msg)
reconciliation_record_id = msg.id
write_result = "not_applicable"
return StateChangeResponse(
target_type=body.target_type,
@@ -99,6 +147,7 @@ async def classify_state_change(
follow_up=classification.follow_up,
write_through_result=write_result,
workplan_path=workplan_ref.relative_path if workplan_ref else None,
reconciliation_record_id=reconciliation_record_id,
)
task = await session.get(Task, body.target_id)
@@ -137,6 +186,7 @@ async def classify_state_change(
blocking_reason=body.blocking_reason,
)
write_result = "not_attempted"
reconciliation_record_id = None
if body.apply:
if (
classification.reconciliation_class == ReconciliationClass.WRITE_THROUGH
@@ -150,6 +200,19 @@ async def classify_state_change(
await session.commit()
write_result = "applied"
else:
msg = _deferred_message(
body=body,
current_status=current_status,
target_status=target_status,
classification=classification.reconciliation_class,
reason=classification.reason,
follow_up=classification.follow_up,
workplan_path=workplan_ref.relative_path if workplan_ref else None,
)
session.add(msg)
await session.commit()
await session.refresh(msg)
reconciliation_record_id = msg.id
write_result = "not_applicable"
return StateChangeResponse(
target_type=body.target_type,
@@ -166,4 +229,5 @@ async def classify_state_change(
follow_up=classification.follow_up,
write_through_result=write_result,
workplan_path=workplan_ref.relative_path if workplan_ref else None,
reconciliation_record_id=reconciliation_record_id,
)

1
api/schemas/reconciliation.py
View File

@@ -39,3 +39,4 @@ class StateChangeResponse(BaseModel):
follow_up: str
write_through_result: Literal["not_attempted", "applied", "not_applicable"] = "not_attempted"
workplan_path: str | None = None
reconciliation_record_id: uuid.UUID | None = None

60
tests/test_routers_core.py
View File

@@ -536,10 +536,19 @@ class TestReconciliationEndpoints:
assert body["file_backed"] is False
assert body["reconciliation_class"] == "deferred"
assert body["write_through_result"] == "not_applicable"
assert body["reconciliation_record_id"]
r = await client.get(f"/workstreams/{ws['id']}")
assert r.json()["status"] == "active"
r = await client.get("/messages/?to_agent=state-hub&unread_only=true")
assert r.status_code == 200
messages = r.json()
assert len(messages) == 1
assert messages[0]["id"] == body["reconciliation_record_id"]
assert "Reconcile workstream state change" in messages[0]["subject"]
assert ws["id"] in messages[0]["body"]
async def test_apply_task_write_through_patches_task_block_then_db(self, client, tmp_path):
await _create_domain(client)
repo_root = tmp_path / "repo"
@@ -587,3 +596,54 @@ class TestReconciliationEndpoints:
r = await client.get(f"/tasks/{task['id']}")
assert r.json()["status"] == "in_progress"
async def test_apply_task_confirmation_case_creates_reconciliation_message(self, client, tmp_path):
await _create_domain(client)
repo_root = tmp_path / "repo"
workplans = repo_root / "workplans"
workplans.mkdir(parents=True)
repo = await _create_repo(client, local_path=repo_root)
topic = await _create_topic(client)
ws = await _create_workstream(client, topic["id"], repo_id=repo["id"])
task = await _create_task(client, ws["id"])
wp = workplans / "STATE-WP-9999-demo.md"
wp.write_text(
"---\n"
"id: STATE-WP-9999\n"
"type: workplan\n"
"title: Demo\n"
"domain: custodian\n"
"repo: state-hub\n"
"status: active\n"
f"state_hub_workstream_id: \"{ws['id']}\"\n"
"---\n\n"
"## Demo Task\n\n"
"```task\n"
"id: STATE-WP-9999-T01\n"
"status: todo\n"
"priority: high\n"
f"state_hub_task_id: \"{task['id']}\"\n"
"```\n",
encoding="utf-8",
)
r = await client.post("/reconciliation/state-change", json={
"target_type": "task",
"target_id": task["id"],
"target_status": "blocked",
"apply": True,
})
assert r.status_code == 200, r.text
body = r.json()
assert body["reconciliation_class"] == "human_confirmation"
assert body["write_through_result"] == "not_applicable"
assert body["reconciliation_record_id"]
r = await client.get(f"/tasks/{task['id']}")
assert r.json()["status"] == "todo"
r = await client.get("/messages/?to_agent=state-hub&unread_only=true")
messages = r.json()
assert len(messages) == 1
assert "blocking reason" in messages[0]["body"]

10
workplans/STATE-WP-0048-ui-state-change-reconciliation.md
View File

@@ -119,7 +119,7 @@ relative workplan path plus `write_through_result: applied`.
```task
id: STATE-WP-0048-T04
status: todo
status: done
priority: high
state_hub_task_id: "2ed06c09-7b92-4b82-bcd3-090a1e320a88"
```
@@ -129,6 +129,11 @@ message, or task that makes the pending file update visible and schedulable.
Done when UI changes that cannot write through are never silent DB-only drift.
Result 2026-05-23: attempted `apply: true` changes that classify as deferred
or human-confirmation now create an unread message for `state-hub` with actor,
intent, target id, current/target status, reason, follow-up, and workplan path
when known.
## T05 - Surface Reconciliation State In Dashboard
```task
@@ -173,6 +178,9 @@ deferred reconciliation.
Done when UI state changes are covered as first-class ADR-001 workflows.
Progress 2026-05-23: added API tests for classify-only responses, safe
write-through, missing-file deferral, and human-confirmation message creation.
## Acceptance Criteria
- Dashboard state changes never create silent DB/file divergence.