LZH-YS1998
886 lines
38 KiB
Python
886 lines
38 KiB
Python
"""Read-model builders for the CLI board."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import time as _time
|
|
from collections import defaultdict
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
from opc.layer2_organization.phase import (
|
|
Phase,
|
|
coerce_phase,
|
|
kanban_column,
|
|
should_hide_work_item_from_company_kanban,
|
|
)
|
|
from opc.layer2_organization.work_item_context_view import WorkItemContextView
|
|
from opc.layer2_organization.work_item_links import task_by_linked_work_item_id
|
|
from opc.presentation.kanban import (
|
|
COMPANY_KANBAN_COLUMNS,
|
|
build_base_task_payload,
|
|
datetime_to_timestamp,
|
|
)
|
|
|
|
from ..state.models import (
|
|
BoardAlert,
|
|
BoardMetrics,
|
|
BoardSnapshot,
|
|
BoardTaskView,
|
|
LinkedExecutionView,
|
|
OrgEmployeeView,
|
|
OrgRoleView,
|
|
OrgSnapshotView,
|
|
PendingCheckpointView,
|
|
PipelineSnapshot,
|
|
PipelineWorkItemView,
|
|
SessionMessageView,
|
|
SessionSummaryView,
|
|
TaskDetailView,
|
|
)
|
|
|
|
if TYPE_CHECKING:
|
|
from .engine_facade import EngineFacade
|
|
|
|
|
|
class BoardRepository:
|
|
"""Builds CLI-specific board and detail snapshots from the main OPC store."""
|
|
|
|
def __init__(self, facade: "EngineFacade", project_id: str | None = None) -> None:
|
|
self.facade = facade
|
|
self.project_id = project_id
|
|
|
|
async def load_snapshot(self) -> BoardSnapshot:
|
|
engine = await self.facade.ensure_ready()
|
|
if not engine.store:
|
|
return BoardSnapshot(project_id=self._project_id)
|
|
|
|
tasks = await engine.store.get_tasks(project_id=self._project_id)
|
|
checkpoints = await engine.store.get_pending_checkpoints(project_id=self._project_id)
|
|
checkpoint_by_session = self._checkpoint_by_session(checkpoints)
|
|
await self._enrich_checkpoint_payloads(checkpoint_by_session)
|
|
|
|
company_snapshot = await self._maybe_build_company_snapshot(
|
|
engine,
|
|
tasks=tasks,
|
|
checkpoint_by_session=checkpoint_by_session,
|
|
checkpoint_count=len(checkpoints),
|
|
)
|
|
if company_snapshot is not None:
|
|
return company_snapshot
|
|
|
|
visible_tasks, hidden_by_origin, hidden_count = self._split_visible_tasks(tasks)
|
|
|
|
task_views: list[BoardTaskView] = []
|
|
for display_num, task in enumerate(visible_tasks, start=1):
|
|
task_views.append(
|
|
self._task_to_view(
|
|
task,
|
|
checkpoint=checkpoint_by_session.get(getattr(task, "session_id", None)),
|
|
linked_tasks=hidden_by_origin.get(getattr(task, "id", ""), []),
|
|
display_num=display_num,
|
|
)
|
|
)
|
|
|
|
session_summaries = [self._session_summary_view(task) for task in task_views]
|
|
alerts = self._alerts_from_tasks(task_views)
|
|
return BoardSnapshot(
|
|
project_id=self._project_id,
|
|
tasks=task_views,
|
|
hidden_task_count=hidden_count,
|
|
pending_checkpoint_count=len(checkpoints),
|
|
session_summaries=session_summaries,
|
|
alerts=alerts,
|
|
metrics=self._build_metrics(task_views, hidden_count=hidden_count, pending_checkpoint_count=len(checkpoints)),
|
|
)
|
|
|
|
async def _maybe_build_company_snapshot(
|
|
self,
|
|
engine: Any,
|
|
*,
|
|
tasks: list[Any],
|
|
checkpoint_by_session: dict[str | None, PendingCheckpointView],
|
|
checkpoint_count: int,
|
|
) -> BoardSnapshot | None:
|
|
"""Return a company-mode snapshot if an active delegation run exists.
|
|
|
|
Cards are sourced from `DelegationWorkItem`. Runtime `Task` objects are
|
|
kept only as audit references (`runtime_task_id` / `session_id`).
|
|
"""
|
|
store = engine.store
|
|
list_runs = getattr(store, "list_open_delegation_runs", None)
|
|
list_items = getattr(store, "list_delegation_work_items", None)
|
|
if list_runs is None or list_items is None:
|
|
return None
|
|
try:
|
|
open_runs = await list_runs(project_id=self._project_id)
|
|
except TypeError:
|
|
open_runs = await list_runs()
|
|
if not open_runs:
|
|
return None
|
|
active_run = open_runs[0]
|
|
run_id = str(getattr(active_run, "run_id", "") or "").strip()
|
|
if not run_id:
|
|
return None
|
|
work_items = await list_items(run_id)
|
|
if not work_items:
|
|
return None
|
|
|
|
hydrate_links = getattr(store, "hydrate_task_work_item_links", None)
|
|
if callable(hydrate_links):
|
|
await hydrate_links(tasks)
|
|
task_by_work_item_id = task_by_linked_work_item_id(tasks)
|
|
|
|
visible_items: list[Any] = []
|
|
hidden_count = 0
|
|
for item in work_items:
|
|
metadata = dict(getattr(item, "metadata", {}) or {})
|
|
if not str(getattr(item, "parent_work_item_id", "") or "").strip():
|
|
# Skip the synthetic root work item — the kanban shows leaf delegations.
|
|
hidden_count += 1
|
|
continue
|
|
if bool(metadata.get("attention_work_item", False)):
|
|
hidden_count += 1
|
|
continue
|
|
if should_hide_work_item_from_company_kanban(metadata):
|
|
hidden_count += 1
|
|
continue
|
|
visible_items.append(item)
|
|
|
|
task_views: list[BoardTaskView] = []
|
|
for display_num, item in enumerate(visible_items, start=1):
|
|
linked_task = task_by_work_item_id.get(str(getattr(item, "work_item_id", "") or "").strip())
|
|
checkpoint = None
|
|
linked_session_id = getattr(linked_task, "session_id", None) if linked_task is not None else None
|
|
if linked_session_id:
|
|
checkpoint = checkpoint_by_session.get(linked_session_id)
|
|
task_views.append(
|
|
self._work_item_to_view(
|
|
item,
|
|
linked_task=linked_task,
|
|
checkpoint=checkpoint,
|
|
display_num=display_num,
|
|
)
|
|
)
|
|
|
|
session_summaries = [self._session_summary_view(task) for task in task_views]
|
|
alerts = self._alerts_from_tasks(task_views)
|
|
column_order = [column.column_id for column in COMPANY_KANBAN_COLUMNS]
|
|
return BoardSnapshot(
|
|
project_id=self._project_id,
|
|
tasks=task_views,
|
|
hidden_task_count=hidden_count,
|
|
pending_checkpoint_count=checkpoint_count,
|
|
session_summaries=session_summaries,
|
|
alerts=alerts,
|
|
metrics=self._build_metrics(
|
|
task_views,
|
|
hidden_count=hidden_count,
|
|
pending_checkpoint_count=checkpoint_count,
|
|
),
|
|
mode="company",
|
|
column_order=column_order,
|
|
)
|
|
|
|
async def load_task_detail(self, task_id: str) -> TaskDetailView | None:
|
|
engine = await self.facade.ensure_ready()
|
|
if not engine.store:
|
|
return None
|
|
|
|
tasks = await engine.store.get_tasks(project_id=self._project_id)
|
|
checkpoints = await engine.store.get_pending_checkpoints(project_id=self._project_id)
|
|
checkpoint_by_session = self._checkpoint_by_session(checkpoints)
|
|
await self._enrich_checkpoint_payloads(checkpoint_by_session)
|
|
|
|
# In company mode the kanban card_id is a DelegationWorkItem.work_item_id,
|
|
# not a Task.id. Try the work-item path first; fall back to the runtime
|
|
# Task path so standard mode (and detail links by Task.id) still works.
|
|
work_item_detail = await self._load_work_item_detail(
|
|
engine,
|
|
work_item_id=task_id,
|
|
tasks=tasks,
|
|
checkpoint_by_session=checkpoint_by_session,
|
|
)
|
|
if work_item_detail is not None:
|
|
return work_item_detail
|
|
|
|
visible_tasks, _, _ = self._split_visible_tasks(tasks)
|
|
target = next((task for task in tasks if getattr(task, "id", "") == task_id), None)
|
|
if target is None:
|
|
return None
|
|
|
|
pending = checkpoint_by_session.get(getattr(target, "session_id", None))
|
|
linked: list[Any] = []
|
|
seen_ids: set[str] = set()
|
|
for task in tasks:
|
|
origin_task_id = self._origin_task_id(task)
|
|
if origin_task_id == task_id and getattr(task, "id", "") != task_id:
|
|
if task.id not in seen_ids:
|
|
linked.append(task)
|
|
seen_ids.add(task.id)
|
|
continue
|
|
if getattr(task, "parent_session_id", None) and getattr(task, "parent_session_id", None) == getattr(target, "session_id", None):
|
|
if task.id not in seen_ids and task.id != task_id:
|
|
linked.append(task)
|
|
seen_ids.add(task.id)
|
|
|
|
transcript: list[SessionMessageView] = []
|
|
if getattr(target, "session_id", None):
|
|
raw_transcript = await engine.store.get_session_transcript(target.session_id)
|
|
transcript = [msg for msg in (self._transcript_item_to_view(item) for item in raw_transcript) if msg is not None]
|
|
|
|
display_num = next(
|
|
(index for index, task in enumerate(visible_tasks, start=1) if getattr(task, "id", "") == task_id),
|
|
0,
|
|
)
|
|
task_view = self._task_to_view(target, checkpoint=pending, linked_tasks=linked, display_num=display_num)
|
|
linked_views = [self._linked_execution_view(task) for task in linked]
|
|
result = getattr(target, "result", None) or {}
|
|
result_content = result.get("content") if isinstance(result, dict) else None
|
|
artifacts = result.get("artifacts", []) if isinstance(result, dict) else []
|
|
metadata = getattr(target, "metadata", {}) if isinstance(getattr(target, "metadata", {}), dict) else {}
|
|
context_preview = (
|
|
metadata.get("handoff_context")
|
|
or metadata.get("context_preview")
|
|
or metadata.get("secretary_context")
|
|
or None
|
|
)
|
|
return TaskDetailView(
|
|
task=task_view,
|
|
transcript=transcript,
|
|
linked_executions=linked_views,
|
|
progress_entries=list(metadata.get("progress_log", []) or []),
|
|
pending_checkpoint=pending,
|
|
result_content=result_content,
|
|
artifacts=artifacts if isinstance(artifacts, list) else [artifacts],
|
|
context_preview=str(context_preview).strip() if context_preview else None,
|
|
)
|
|
|
|
async def _load_work_item_detail(
|
|
self,
|
|
engine: Any,
|
|
*,
|
|
work_item_id: str,
|
|
tasks: list[Any],
|
|
checkpoint_by_session: dict[str | None, PendingCheckpointView],
|
|
) -> TaskDetailView | None:
|
|
"""Build a detail view keyed by DelegationWorkItem.work_item_id.
|
|
|
|
Returns None when the id is not a known work item (caller falls back
|
|
to the Task-id path).
|
|
"""
|
|
store = engine.store
|
|
list_runs = getattr(store, "list_open_delegation_runs", None)
|
|
list_items = getattr(store, "list_delegation_work_items", None)
|
|
if list_runs is None or list_items is None or not work_item_id:
|
|
return None
|
|
try:
|
|
open_runs = await list_runs(project_id=self._project_id)
|
|
except TypeError:
|
|
open_runs = await list_runs()
|
|
if not open_runs:
|
|
return None
|
|
target_item: Any | None = None
|
|
for run in open_runs:
|
|
run_id = str(getattr(run, "run_id", "") or "").strip()
|
|
if not run_id:
|
|
continue
|
|
for item in await list_items(run_id):
|
|
if str(getattr(item, "work_item_id", "") or "").strip() == work_item_id:
|
|
target_item = item
|
|
break
|
|
if target_item is not None:
|
|
break
|
|
if target_item is None:
|
|
return None
|
|
|
|
task_by_work_item_id = task_by_linked_work_item_id(tasks)
|
|
linked_task = task_by_work_item_id.get(work_item_id)
|
|
|
|
linked_session_id = str(getattr(linked_task, "session_id", "") or "").strip() if linked_task is not None else ""
|
|
checkpoint = checkpoint_by_session.get(linked_session_id) if linked_session_id else None
|
|
|
|
transcript: list[SessionMessageView] = []
|
|
if linked_session_id:
|
|
raw_transcript = await engine.store.get_session_transcript(linked_session_id)
|
|
transcript = [
|
|
msg
|
|
for msg in (self._transcript_item_to_view(item) for item in raw_transcript)
|
|
if msg is not None
|
|
]
|
|
|
|
task_view = self._work_item_to_view(
|
|
target_item,
|
|
linked_task=linked_task,
|
|
checkpoint=checkpoint,
|
|
display_num=0,
|
|
)
|
|
result = getattr(linked_task, "result", None) or {} if linked_task is not None else {}
|
|
result_content = result.get("content") if isinstance(result, dict) else None
|
|
artifacts = result.get("artifacts", []) if isinstance(result, dict) else []
|
|
# Prefer work-item handoff context; fall back to runtime Task metadata.
|
|
linked_metadata = (
|
|
getattr(linked_task, "metadata", None) or {}
|
|
if linked_task is not None
|
|
else {}
|
|
)
|
|
view = WorkItemContextView(target_item, linked_task)
|
|
context_preview = (
|
|
view.get("handoff_context")
|
|
or view.get("context_preview")
|
|
or (linked_metadata.get("secretary_context") if isinstance(linked_metadata, dict) else None)
|
|
or None
|
|
)
|
|
progress_log = view.get_list("progress_log")
|
|
return TaskDetailView(
|
|
task=task_view,
|
|
transcript=transcript,
|
|
linked_executions=[self._linked_execution_view(linked_task)] if linked_task is not None else [],
|
|
progress_entries=progress_log,
|
|
pending_checkpoint=checkpoint,
|
|
result_content=result_content,
|
|
artifacts=artifacts if isinstance(artifacts, list) else [artifacts],
|
|
context_preview=str(context_preview).strip() if context_preview else None,
|
|
)
|
|
|
|
async def load_pipeline_state(
|
|
self,
|
|
parent_task_id: str,
|
|
runtime_lookup: dict[str, Any] | None = None,
|
|
) -> PipelineSnapshot | None:
|
|
"""Build the terminal Office-style work-item pipeline for a run."""
|
|
engine = await self.facade.ensure_ready()
|
|
store = getattr(engine, "store", None)
|
|
if not store:
|
|
return None
|
|
list_runs = getattr(store, "list_open_delegation_runs", None)
|
|
list_items = getattr(store, "list_delegation_work_items", None)
|
|
if list_runs is None or list_items is None:
|
|
return None
|
|
try:
|
|
runs = list(await list_runs(project_id=self._project_id))
|
|
except TypeError:
|
|
runs = list(await list_runs())
|
|
if not runs:
|
|
return None
|
|
|
|
tasks = await store.get_tasks(project_id=self._project_id) if hasattr(store, "get_tasks") else []
|
|
hydrate_links = getattr(store, "hydrate_task_work_item_links", None)
|
|
if callable(hydrate_links):
|
|
await hydrate_links(tasks)
|
|
linked_tasks = task_by_linked_work_item_id(tasks)
|
|
|
|
selected_run = None
|
|
selected_items: list[Any] = []
|
|
target = str(parent_task_id or "").strip()
|
|
for run in runs:
|
|
run_id = str(getattr(run, "run_id", "") or "").strip()
|
|
items = list(await list_items(run_id)) if run_id else []
|
|
run_keys = {
|
|
run_id,
|
|
str(getattr(run, "parent_task_id", "") or "").strip(),
|
|
str(getattr(run, "task_id", "") or "").strip(),
|
|
str(getattr(run, "parent_session_id", "") or "").strip(),
|
|
}
|
|
item_keys = {str(getattr(item, "work_item_id", "") or "").strip() for item in items}
|
|
if not target or target in run_keys or target in item_keys:
|
|
selected_run = run
|
|
selected_items = items
|
|
break
|
|
if selected_run is None:
|
|
selected_run = runs[0]
|
|
run_id = str(getattr(selected_run, "run_id", "") or "").strip()
|
|
selected_items = list(await list_items(run_id)) if run_id else []
|
|
if not selected_items:
|
|
return None
|
|
|
|
work_items: list[PipelineWorkItemView] = []
|
|
start_ts = 0.0
|
|
end_ts = 0.0
|
|
for item in selected_items:
|
|
work_item_id = str(getattr(item, "work_item_id", "") or "").strip()
|
|
linked_task = linked_tasks.get(work_item_id)
|
|
linked_task_id = str(getattr(linked_task, "id", "") or "").strip() if linked_task is not None else ""
|
|
metadata = dict(getattr(item, "metadata", {}) or {})
|
|
linked_metadata = dict(getattr(linked_task, "metadata", {}) or {}) if linked_task is not None else {}
|
|
phase = coerce_phase(getattr(item, "phase", Phase.READY))
|
|
created_at = datetime_to_timestamp(getattr(item, "created_at", None))
|
|
updated_at = datetime_to_timestamp(getattr(item, "updated_at", None) or getattr(item, "created_at", None))
|
|
if created_at and (not start_ts or created_at < start_ts):
|
|
start_ts = created_at
|
|
if updated_at and updated_at > end_ts:
|
|
end_ts = updated_at
|
|
runtime = None
|
|
if runtime_lookup:
|
|
runtime = runtime_lookup.get(linked_task_id) or runtime_lookup.get(work_item_id)
|
|
work_items.append(PipelineWorkItemView(
|
|
projection_id=work_item_id,
|
|
title=str(getattr(item, "title", "") or work_item_id),
|
|
role_id=str(getattr(item, "role_id", "") or ""),
|
|
status=phase.value,
|
|
assigned_to=str(getattr(linked_task, "assigned_to", "") or getattr(item, "role_id", "") or ""),
|
|
task_id=linked_task_id or None,
|
|
runtime_task_id=linked_task_id or None,
|
|
execution_turn_id=linked_task_id or None,
|
|
session_id=str(getattr(linked_task, "session_id", "") or "") or None,
|
|
elapsed_sec=max(0.0, (updated_at or _time.time()) - created_at) if created_at else 0.0,
|
|
current_tool=getattr(runtime, "current_tool", None) if runtime is not None else None,
|
|
tool_elapsed_ms=int(getattr(runtime, "tool_elapsed_ms", 0) or 0) if runtime is not None else 0,
|
|
last_tool_summary=str(getattr(runtime, "last_tool_summary", "") or "") if runtime is not None else "",
|
|
context_remaining_pct=int(getattr(runtime, "context_remaining_pct", 0) or 0) if runtime is not None else 0,
|
|
turn_cost_usd=float(getattr(runtime, "turn_cost_usd", 0.0) or 0.0) if runtime is not None else 0.0,
|
|
has_gate=bool(metadata.get("gate_type") or linked_metadata.get("checkpoint_hint") or (getattr(runtime, "pending_permission_count", 0) if runtime is not None else 0)),
|
|
gate_type=str(metadata.get("gate_type") or linked_metadata.get("checkpoint_type") or "") or None,
|
|
dependencies=[
|
|
str(dep).strip()
|
|
for dep in list(metadata.get("dependency_work_item_ids", []) or [])
|
|
if str(dep).strip()
|
|
],
|
|
parallel_group=str(metadata.get("parallel_group", "") or "") or None,
|
|
))
|
|
|
|
done_count = sum(1 for item in work_items if item.status in {"done", "reviewed", "delivered"})
|
|
parent_title = (
|
|
str(getattr(selected_run, "title", "") or "")
|
|
or str(getattr(selected_run, "summary", "") or "")
|
|
or str(getattr(selected_run, "run_id", "") or "")
|
|
)
|
|
return PipelineSnapshot(
|
|
parent_task_id=str(getattr(selected_run, "parent_task_id", "") or getattr(selected_run, "task_id", "") or target),
|
|
parent_title=parent_title,
|
|
profile=str(getattr(selected_run, "profile", "") or getattr(selected_run, "company_profile", "") or ""),
|
|
work_items=work_items,
|
|
done_count=done_count,
|
|
total_count=len(work_items),
|
|
elapsed_sec=max(0.0, (end_ts or _time.time()) - start_ts) if start_ts else 0.0,
|
|
)
|
|
|
|
async def load_org_snapshot(self) -> OrgSnapshotView | None:
|
|
"""Build a read-only view of the current org structure."""
|
|
engine = await self.facade.ensure_ready()
|
|
if not engine.org_engine:
|
|
return None
|
|
|
|
org = engine.org_engine
|
|
|
|
# Roles as tree
|
|
all_agents = org.list_agents()
|
|
all_employees = org.list_employees()
|
|
emp_count_by_role: dict[str, int] = {}
|
|
for emp in all_employees:
|
|
emp_count_by_role[emp.role_id] = emp_count_by_role.get(emp.role_id, 0) + 1
|
|
|
|
try:
|
|
tree_raw = org.get_org_tree()
|
|
except Exception:
|
|
tree_raw = []
|
|
|
|
def _build_role_tree(nodes: list[dict]) -> list[OrgRoleView]:
|
|
result: list[OrgRoleView] = []
|
|
for node in nodes:
|
|
agent = node.get("agent")
|
|
if agent is None:
|
|
continue
|
|
role_id = getattr(agent, "role_id", "")
|
|
result.append(OrgRoleView(
|
|
role_id=role_id,
|
|
name=getattr(agent, "name", role_id),
|
|
responsibility=getattr(agent, "responsibility", ""),
|
|
reports_to=getattr(agent, "reports_to", "owner"),
|
|
employee_count=emp_count_by_role.get(role_id, 0),
|
|
children=_build_role_tree(node.get("reports", [])),
|
|
))
|
|
return result
|
|
|
|
role_tree = _build_role_tree(tree_raw)
|
|
|
|
# Employees
|
|
employee_views = [
|
|
OrgEmployeeView(
|
|
employee_id=emp.employee_id,
|
|
name=emp.name,
|
|
role_id=emp.role_id,
|
|
category=emp.category,
|
|
domains=list(emp.domains),
|
|
seniority=emp.seniority,
|
|
)
|
|
for emp in all_employees
|
|
]
|
|
|
|
profile = org.get_company_profile()
|
|
|
|
return OrgSnapshotView(
|
|
role_tree=role_tree,
|
|
employees=employee_views,
|
|
company_profile=profile,
|
|
role_count=len(all_agents),
|
|
employee_count=len(all_employees),
|
|
)
|
|
|
|
@property
|
|
def _project_id(self) -> str:
|
|
return self.project_id or self.facade.project_id or "default"
|
|
|
|
@staticmethod
|
|
def _origin_task_id(task: Any) -> str | None:
|
|
metadata = getattr(task, "metadata", None)
|
|
if not isinstance(metadata, dict):
|
|
return None
|
|
origin_task_id = str(metadata.get("origin_task_id", "") or "").strip()
|
|
return origin_task_id or None
|
|
|
|
def _split_visible_tasks(self, tasks: list[Any]) -> tuple[list[Any], dict[str, list[Any]], int]:
|
|
visible_tasks: list[Any] = []
|
|
hidden_by_origin: dict[str, list[Any]] = defaultdict(list)
|
|
hidden_count = 0
|
|
for task in tasks:
|
|
origin_task_id = self._origin_task_id(task)
|
|
if origin_task_id and origin_task_id != getattr(task, "id", ""):
|
|
hidden_by_origin[origin_task_id].append(task)
|
|
hidden_count += 1
|
|
continue
|
|
visible_tasks.append(task)
|
|
return visible_tasks, hidden_by_origin, hidden_count
|
|
|
|
@staticmethod
|
|
def _checkpoint_prompt(checkpoint: Any) -> tuple[str, str]:
|
|
payload = getattr(checkpoint, "payload", {}) or {}
|
|
prompt = (
|
|
payload.get("prompt")
|
|
or payload.get("message")
|
|
or payload.get("summary")
|
|
or payload.get("original_message")
|
|
or ""
|
|
)
|
|
prompt_text = str(prompt).strip()
|
|
if not prompt_text:
|
|
prompt_text = json.dumps(payload, ensure_ascii=False, indent=2)[:600]
|
|
|
|
summary = (
|
|
payload.get("feedback_scope")
|
|
or payload.get("work_item_projection_title")
|
|
or payload.get("summary")
|
|
or payload.get("title")
|
|
or getattr(checkpoint, "checkpoint_type", "pending")
|
|
)
|
|
return str(summary).strip(), prompt_text
|
|
|
|
def _checkpoint_view(self, checkpoint: Any) -> PendingCheckpointView:
|
|
summary, prompt = self._checkpoint_prompt(checkpoint)
|
|
return PendingCheckpointView(
|
|
checkpoint_id=str(getattr(checkpoint, "checkpoint_id", "") or ""),
|
|
checkpoint_type=str(getattr(checkpoint, "checkpoint_type", "") or ""),
|
|
status=str(getattr(checkpoint, "status", "") or ""),
|
|
session_id=getattr(checkpoint, "session_id", None),
|
|
task_id=getattr(checkpoint, "task_id", None),
|
|
summary=summary,
|
|
prompt=prompt,
|
|
payload=dict(getattr(checkpoint, "payload", {}) or {}),
|
|
)
|
|
|
|
def _checkpoint_by_session(self, checkpoints: list[Any]) -> dict[str | None, PendingCheckpointView]:
|
|
result: dict[str | None, PendingCheckpointView] = {}
|
|
for checkpoint in checkpoints:
|
|
session_id = getattr(checkpoint, "session_id", None)
|
|
if session_id not in result:
|
|
result[session_id] = self._checkpoint_view(checkpoint)
|
|
return result
|
|
|
|
async def _enrich_checkpoint_payloads(
|
|
self,
|
|
checkpoint_map: dict[str | None, PendingCheckpointView],
|
|
) -> None:
|
|
"""Enrich reorg checkpoint payloads with full proposal data from store."""
|
|
engine = await self.facade.ensure_ready()
|
|
if not engine.store:
|
|
return
|
|
for view in checkpoint_map.values():
|
|
if view.checkpoint_type != "company_reorg_pending":
|
|
continue
|
|
proposal_id = str(view.payload.get("proposal_id", "") or "").strip()
|
|
if not proposal_id:
|
|
continue
|
|
try:
|
|
proposal = await engine.store.get_reorg_proposal(proposal_id)
|
|
except Exception:
|
|
continue
|
|
if proposal is None:
|
|
continue
|
|
view.payload["title"] = proposal.title
|
|
view.payload["scope"] = proposal.scope.value if hasattr(proposal.scope, "value") else str(proposal.scope)
|
|
view.payload["risk_level"] = proposal.risk_level.value if hasattr(proposal.risk_level, "value") else str(proposal.risk_level)
|
|
view.payload["summary"] = proposal.summary
|
|
view.payload["rationale"] = proposal.rationale
|
|
view.payload["impact_summary"] = dict(proposal.impact_summary) if proposal.impact_summary else {}
|
|
changeset = proposal.changeset
|
|
if changeset:
|
|
view.payload["role_changes"] = [
|
|
{"action": rc.action, "role_id": rc.role_id, "replacement_role_id": getattr(rc, "replacement_role_id", ""), "reason": getattr(rc, "reason", "")}
|
|
for rc in (changeset.role_changes if hasattr(changeset, "role_changes") else [])
|
|
]
|
|
|
|
def _work_item_to_view(
|
|
self,
|
|
item: Any,
|
|
*,
|
|
linked_task: Any | None,
|
|
checkpoint: PendingCheckpointView | None,
|
|
display_num: int = 0,
|
|
) -> BoardTaskView:
|
|
"""Build a BoardTaskView from a DelegationWorkItem (company-mode card).
|
|
|
|
The card identity is the work_item_id; runtime Task / session are
|
|
kept only as audit references via runtime_task_id / session_id.
|
|
"""
|
|
normalized_metadata = dict(getattr(item, "metadata", {}) or {})
|
|
phase = coerce_phase(getattr(item, "phase", Phase.READY))
|
|
column_id = kanban_column(phase).replace("_", "-")
|
|
canonical_status = phase.value
|
|
work_item_id = str(getattr(item, "work_item_id", "") or "").strip()
|
|
role_id = str(getattr(item, "role_id", "") or "").strip()
|
|
title = str(getattr(item, "title", "") or "").strip()
|
|
summary = str(getattr(item, "summary", "") or "").strip()
|
|
kind = str(getattr(item, "kind", "") or "").strip()
|
|
dependencies = [
|
|
str(dep).strip()
|
|
for dep in list(normalized_metadata.get("dependency_work_item_ids", []) or [])
|
|
if str(dep).strip()
|
|
]
|
|
created_at = datetime_to_timestamp(getattr(item, "created_at", None))
|
|
updated_at = datetime_to_timestamp(getattr(item, "updated_at", None) or getattr(item, "created_at", None))
|
|
|
|
# Audit/back-references (NOT used to drive lifecycle on the card).
|
|
linked_task_id = str(getattr(linked_task, "id", "") or "").strip() if linked_task is not None else ""
|
|
linked_session_id = str(getattr(linked_task, "session_id", "") or "").strip() if linked_task is not None else ""
|
|
result = getattr(linked_task, "result", None) or {} if linked_task is not None else {}
|
|
result_content = result.get("content") if isinstance(result, dict) else None
|
|
artifacts = result.get("artifacts", []) if isinstance(result, dict) else []
|
|
# Merge work-item metadata first, then linked-task metadata so the card
|
|
# surfaces work-item-truth (status/dependencies) plus runtime telemetry.
|
|
merged_metadata: dict[str, Any] = dict(normalized_metadata)
|
|
if linked_task is not None:
|
|
linked_meta = getattr(linked_task, "metadata", None) or {}
|
|
if isinstance(linked_meta, dict):
|
|
for key, value in linked_meta.items():
|
|
merged_metadata.setdefault(key, value)
|
|
|
|
return BoardTaskView(
|
|
task_id=work_item_id,
|
|
title=title,
|
|
description=summary,
|
|
status=canonical_status,
|
|
column_id=column_id,
|
|
priority=None,
|
|
assignee_ids=[role_id] if role_id else [],
|
|
assigned_to=role_id,
|
|
tags=[kind] if kind else [],
|
|
session_id=linked_session_id or None,
|
|
created_at=created_at,
|
|
updated_at=updated_at,
|
|
metadata=merged_metadata,
|
|
pending_checkpoint=checkpoint,
|
|
linked_task_count=1 if linked_task is not None else 0,
|
|
result_content=result_content,
|
|
artifacts=artifacts if isinstance(artifacts, list) else [artifacts],
|
|
origin_task_id=None,
|
|
display_id=f"OPC-{display_num}" if display_num else "",
|
|
dependencies=dependencies,
|
|
work_item_id=work_item_id,
|
|
runtime_task_id=linked_task_id or None,
|
|
execution_turn_id=linked_task_id or None,
|
|
)
|
|
|
|
def _task_to_view(
|
|
self,
|
|
task: Any,
|
|
*,
|
|
checkpoint: PendingCheckpointView | None,
|
|
linked_tasks: list[Any],
|
|
display_num: int = 0,
|
|
) -> BoardTaskView:
|
|
payload = build_base_task_payload(task, display_num)
|
|
result = getattr(task, "result", None) or {}
|
|
result_content = result.get("content") if isinstance(result, dict) else None
|
|
artifacts = result.get("artifacts", []) if isinstance(result, dict) else []
|
|
metadata = getattr(task, "metadata", {}) if isinstance(getattr(task, "metadata", {}), dict) else {}
|
|
return BoardTaskView(
|
|
task_id=payload["task_id"],
|
|
title=payload["title"],
|
|
description=payload["description"],
|
|
status=payload["status"],
|
|
column_id=payload["column_id"],
|
|
priority=payload["priority"],
|
|
assignee_ids=payload["assignee_ids"],
|
|
assigned_to=str(getattr(task, "assigned_to", "") or ""),
|
|
tags=payload["tags"],
|
|
session_id=payload["session_id"],
|
|
created_at=float(payload["created_at"]),
|
|
updated_at=float(payload["updated_at"]),
|
|
metadata=metadata,
|
|
pending_checkpoint=checkpoint,
|
|
linked_task_count=len(linked_tasks),
|
|
result_content=result_content,
|
|
artifacts=artifacts if isinstance(artifacts, list) else [artifacts],
|
|
origin_task_id=self._origin_task_id(task),
|
|
display_id=str(payload.get("display_id", "") or ""),
|
|
dependencies=list(payload.get("dependencies", []) or []),
|
|
runtime_task_id=payload["task_id"],
|
|
execution_turn_id=payload["task_id"],
|
|
)
|
|
|
|
@staticmethod
|
|
def _session_summary_view(task: BoardTaskView) -> SessionSummaryView:
|
|
return SessionSummaryView(
|
|
task_id=task.task_id,
|
|
title=task.title,
|
|
status=task.status,
|
|
column_id=task.column_id,
|
|
session_id=task.session_id,
|
|
updated_at=float(task.updated_at),
|
|
created_at=float(task.created_at),
|
|
assigned_to=task.assigned_to,
|
|
priority=task.priority,
|
|
pending_checkpoint=task.pending_checkpoint is not None,
|
|
linked_task_count=task.linked_task_count,
|
|
tags=list(task.tags),
|
|
runtime_task_id=task.runtime_task_id,
|
|
execution_turn_id=task.execution_turn_id or task.runtime_task_id,
|
|
)
|
|
|
|
@staticmethod
|
|
def _alerts_from_tasks(tasks: list[BoardTaskView]) -> list[BoardAlert]:
|
|
alerts: list[BoardAlert] = []
|
|
for task in tasks:
|
|
item_label = "Work item" if task.work_item_id else "Task"
|
|
if task.pending_checkpoint:
|
|
alerts.append(
|
|
BoardAlert(
|
|
alert_id=f"checkpoint:{task.task_id}",
|
|
level="warn",
|
|
title="Checkpoint pending",
|
|
message=f"{task.title} is waiting for human feedback.",
|
|
task_id=task.task_id,
|
|
created_at=float(task.updated_at or task.created_at),
|
|
)
|
|
)
|
|
if task.status in {"blocked", "awaiting_peer", "awaiting_review"}:
|
|
alerts.append(
|
|
BoardAlert(
|
|
alert_id=f"blocked:{task.task_id}",
|
|
level="warn",
|
|
title=f"{item_label} paused",
|
|
message=f"{task.title} is paused in `{task.status}`.",
|
|
task_id=task.task_id,
|
|
created_at=float(task.updated_at or task.created_at),
|
|
)
|
|
)
|
|
if task.status in {"failed", "cancelled"}:
|
|
alerts.append(
|
|
BoardAlert(
|
|
alert_id=f"terminal:{task.task_id}",
|
|
level="error",
|
|
title=f"{item_label} ended abnormally",
|
|
message=f"{task.title} finished with status `{task.status}`.",
|
|
task_id=task.task_id,
|
|
created_at=float(task.updated_at or task.created_at),
|
|
)
|
|
)
|
|
level_weight = {"error": 0, "warn": 1, "info": 2}
|
|
alerts.sort(key=lambda alert: (level_weight.get(alert.level, 99), -float(alert.created_at)))
|
|
return alerts[:12]
|
|
|
|
@staticmethod
|
|
def _build_metrics(
|
|
tasks: list[BoardTaskView],
|
|
*,
|
|
hidden_count: int,
|
|
pending_checkpoint_count: int,
|
|
) -> BoardMetrics:
|
|
todo_count = sum(1 for task in tasks if task.column_id == "todo")
|
|
in_progress_count = sum(1 for task in tasks if task.column_id == "in-progress")
|
|
in_review_count = sum(1 for task in tasks if task.column_id == "in-review")
|
|
done_count = sum(1 for task in tasks if task.column_id == "done")
|
|
blocked_count = sum(1 for task in tasks if task.status in {"blocked", "awaiting_peer", "awaiting_review"})
|
|
failed_count = sum(1 for task in tasks if task.status in {"failed", "cancelled"})
|
|
active_session_count = sum(1 for task in tasks if task.session_id)
|
|
running_count = sum(
|
|
1
|
|
for task in tasks
|
|
if task.status in {"running", "idle", "blocked", "awaiting_peer", "awaiting_review"}
|
|
)
|
|
return BoardMetrics(
|
|
total_tasks=len(tasks) + hidden_count,
|
|
visible_tasks=len(tasks),
|
|
filtered_tasks=len(tasks),
|
|
hidden_task_count=hidden_count,
|
|
todo_count=todo_count,
|
|
in_progress_count=in_progress_count,
|
|
in_review_count=in_review_count,
|
|
done_count=done_count,
|
|
running_count=running_count,
|
|
blocked_count=blocked_count,
|
|
failed_count=failed_count,
|
|
pending_checkpoint_count=pending_checkpoint_count,
|
|
active_session_count=active_session_count,
|
|
alert_count=len(BoardRepository._alerts_from_tasks(tasks)),
|
|
)
|
|
|
|
@staticmethod
|
|
def _linked_execution_view(task: Any) -> LinkedExecutionView:
|
|
payload = build_base_task_payload(task, 0)
|
|
return LinkedExecutionView(
|
|
task_id=payload["task_id"],
|
|
title=payload["title"],
|
|
status=payload["status"],
|
|
assigned_to=str(getattr(task, "assigned_to", "") or ""),
|
|
session_id=payload["session_id"],
|
|
created_at=float(payload["created_at"]),
|
|
updated_at=float(payload["updated_at"]),
|
|
runtime_task_id=payload["task_id"],
|
|
execution_turn_id=payload["task_id"],
|
|
metadata=dict(getattr(task, "metadata", {}) or {}),
|
|
)
|
|
|
|
@staticmethod
|
|
def _render_parts(parts: list[Any]) -> str:
|
|
lines: list[str] = []
|
|
for part in parts:
|
|
part_type = getattr(part, "part_type", "")
|
|
payload = getattr(part, "payload", {}) if isinstance(getattr(part, "payload", {}), dict) else {}
|
|
if part_type == "text":
|
|
text = payload.get("text", "")
|
|
if text:
|
|
lines.append(str(text))
|
|
elif part_type in {"subtask_result", "task_result"}:
|
|
title = payload.get("task_title", "Task")
|
|
summary = payload.get("summary", "")
|
|
lines.append(f"{title}: {summary}".strip(": "))
|
|
return "\n".join(lines).strip()
|
|
|
|
def _transcript_item_to_view(self, item: dict[str, Any]) -> SessionMessageView | None:
|
|
message = item.get("message")
|
|
if not message or getattr(message, "summary_flag", False):
|
|
return None
|
|
|
|
content = self._render_parts(item.get("parts", []))
|
|
if not content:
|
|
return None
|
|
|
|
role = str(getattr(message, "role", "") or "").strip().lower()
|
|
agent_id = str(getattr(message, "agent_id", "") or "").strip()
|
|
sender_name = {
|
|
"user": "You",
|
|
"assistant": "OPC",
|
|
"system": "System",
|
|
"subagent": agent_id.replace("_", " ").replace("-", " ").title() if agent_id else "Subagent",
|
|
}.get(role, agent_id or role.title() or "OPC")
|
|
created_at = getattr(message, "created_at", None)
|
|
timestamp = created_at.timestamp() if hasattr(created_at, "timestamp") else 0.0
|
|
return SessionMessageView(
|
|
message_id=str(getattr(message, "message_id", "") or ""),
|
|
role=role or "assistant",
|
|
sender_name=sender_name,
|
|
content=content,
|
|
created_at=timestamp,
|
|
metadata={"agent_id": agent_id} if agent_id else {},
|
|
)
|