LZH-YS1998
998 lines
52 KiB
Python
998 lines
52 KiB
Python
"""Organization service.
|
|
|
|
The heavy organization mutation implementation still lives in the underlying
|
|
org engine/config models; this service exposes the shared entrypoint used by UI
|
|
and CLI surfaces.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from opc.core.models import normalize_role_runtime_status
|
|
from opc.core.org_config import (
|
|
RunnableOrgConfigError,
|
|
apply_org_config_payload_to_config,
|
|
allocate_org_config_id,
|
|
build_org_config_payload_from_config,
|
|
list_org_config_paths,
|
|
load_org_config_payload,
|
|
org_config_relative_path,
|
|
org_config_path,
|
|
validate_runnable_org_config,
|
|
validate_saved_org_id,
|
|
write_org_config_payload,
|
|
write_org_index,
|
|
)
|
|
from opc.layer2_organization.org_work_item_planner import build_custom_org_work_item_blueprint
|
|
from opc.layer2_organization.phase import kanban_column, should_hide_work_item_from_company_kanban
|
|
from opc.layer2_organization.work_item_identity import (
|
|
work_item_identity_payload,
|
|
work_item_projection_id_from_metadata,
|
|
work_item_turn_type_from_metadata,
|
|
)
|
|
from opc.layer4_tools.output_budget import clip_text
|
|
from opc.plugins.office_ui.org_architecture_snapshot import (
|
|
apply_org_architecture_snapshot,
|
|
build_org_architecture_snapshot,
|
|
dump_org_architecture_snapshot,
|
|
parse_org_architecture_snapshot,
|
|
)
|
|
|
|
from .context import OfficeServiceContext
|
|
from .models import ServiceError, ServiceEvent, ServiceResult
|
|
|
|
|
|
class OrgService:
|
|
def __init__(self, context: OfficeServiceContext) -> None:
|
|
self.context = context
|
|
|
|
def _ensure_custom_org_editable(self) -> None:
|
|
if not self.context.is_custom_org_editable():
|
|
raise ServiceError(
|
|
"org_read_only",
|
|
"Corporate organization is read-only. Select or create a saved custom org before editing.",
|
|
)
|
|
|
|
async def info(self, *, include_events: bool = False) -> ServiceResult:
|
|
"""Build the full Office UI org_info payload.
|
|
|
|
This mirrors the historical WS payload shape so the frontend, CLI, and
|
|
board can share one source of truth without protocol changes.
|
|
"""
|
|
engine = self.context.engine
|
|
result: dict[str, Any] = {
|
|
"roles": [],
|
|
"employees": [],
|
|
"company_profile": "",
|
|
"organization_id": "",
|
|
"organization_name": "",
|
|
"organization_config_file": "",
|
|
"final_decider_role_id": None,
|
|
"top_level_role_ids": [],
|
|
"channels": [],
|
|
"connectors": [],
|
|
"runtime_teams": [],
|
|
"runtime_seats": [],
|
|
"work_items": [],
|
|
"frontier": {},
|
|
"runtime_topology_preview": {},
|
|
"work_item_runtime_preview": {},
|
|
"project_run": {},
|
|
"project_dossier": {},
|
|
"seat_digests": [],
|
|
"revision_links": [],
|
|
"project_recovery": {},
|
|
"org_version": 0,
|
|
"runtime_topology_version": 0,
|
|
}
|
|
cfg_org = getattr(getattr(engine, "config", None), "org", None)
|
|
if cfg_org is not None:
|
|
result["organization_id"] = str(getattr(cfg_org, "organization_id", "") or "")
|
|
result["organization_name"] = str(getattr(cfg_org, "organization_name", "") or "")
|
|
result["organization_config_file"] = str(getattr(cfg_org, "organization_config_file", "") or "")
|
|
|
|
org = getattr(engine, "org_engine", None)
|
|
agents: list[Any] = []
|
|
if org:
|
|
try:
|
|
agents = list(org.list_agents())
|
|
builtin_ids: set[str] = set()
|
|
if self.context.mode_state.exec_mode not in {"org", "custom"}:
|
|
try:
|
|
from opc.layer2_organization.company_runtime_profiles import get_builtin_roles
|
|
|
|
for profile in ("corporate",):
|
|
builtin_ids.update(role.id for role in get_builtin_roles(profile))
|
|
except Exception:
|
|
pass
|
|
result["roles"] = [
|
|
{
|
|
"role_id": agent.role_id,
|
|
"name": agent.name,
|
|
"responsibility": agent.responsibility,
|
|
"status": agent.status.value if hasattr(agent.status, "value") else str(agent.status),
|
|
"reports_to": agent.reports_to,
|
|
"icon": getattr(agent, "icon", None),
|
|
"can_spawn": list(agent.can_spawn) if agent.can_spawn else [],
|
|
"tools": list(agent.tools) if agent.tools else [],
|
|
"is_builtin": agent.role_id in builtin_ids,
|
|
"execution_strategy": agent.runtime_policy.get("execution_strategy", "auto")
|
|
if isinstance(agent.runtime_policy, dict)
|
|
else "auto",
|
|
"preferred_external_agent": agent.preferred_external_agent,
|
|
"prompt_refs": list(agent.prompt_refs) if agent.prompt_refs else [],
|
|
}
|
|
for agent in agents
|
|
]
|
|
except Exception:
|
|
pass
|
|
try:
|
|
employees = list(org.list_employees())
|
|
effective_role_ids = {agent.role_id for agent in agents}
|
|
role_getter = getattr(org, "employee_role_ids", None)
|
|
filtered_employees = []
|
|
for employee in employees:
|
|
if callable(role_getter):
|
|
try:
|
|
role_ids = list(role_getter(employee))
|
|
except Exception:
|
|
role_ids = []
|
|
else:
|
|
role_ids = [str(getattr(employee, "role_id", "") or "").strip()]
|
|
if any(role_id in effective_role_ids for role_id in role_ids):
|
|
filtered_employees.append(employee)
|
|
employees = filtered_employees
|
|
emp_agent_map = {}
|
|
if self.context.agent_store is not None:
|
|
getter = getattr(self.context.agent_store, "get_employee_agent_map", None)
|
|
if callable(getter):
|
|
emp_agent_map = await getter()
|
|
emp_list = []
|
|
for employee in employees:
|
|
emp_meta = dict(getattr(employee, "metadata", {}) or {})
|
|
role_ids: list[str] = []
|
|
role_getter = getattr(org, "employee_role_ids", None)
|
|
if callable(role_getter):
|
|
try:
|
|
role_ids = list(role_getter(employee))
|
|
except Exception:
|
|
role_ids = []
|
|
if not role_ids:
|
|
for value in [
|
|
getattr(employee, "role_id", ""),
|
|
emp_meta.get("home_role_id"),
|
|
*list(emp_meta.get("home_role_ids", []) or []),
|
|
*list(emp_meta.get("staffed_role_ids", []) or []),
|
|
]:
|
|
role_id = str(value or "").strip()
|
|
if role_id and role_id not in role_ids:
|
|
role_ids.append(role_id)
|
|
emp_dict: dict[str, Any] = {
|
|
"employee_id": employee.employee_id,
|
|
"name": employee.name,
|
|
"role_id": employee.role_id,
|
|
"role_ids": role_ids,
|
|
"category": getattr(employee, "category", ""),
|
|
"domains": list(getattr(employee, "domains", [])),
|
|
"seniority": getattr(employee, "seniority", "junior"),
|
|
"status": getattr(employee, "status", "active"),
|
|
"tags": list(getattr(employee, "tags", [])),
|
|
"prompt_refs": list(getattr(employee, "prompt_refs", [])),
|
|
"skill_refs": list(getattr(employee, "skill_refs", [])),
|
|
"preferred_external_agent": getattr(employee, "preferred_external_agent", None),
|
|
"experience_score": 0.0,
|
|
"learned_skill_refs": [],
|
|
"is_default_employee": bool(emp_meta.get("is_default_employee", False)),
|
|
}
|
|
linked = emp_agent_map.get(employee.employee_id)
|
|
if linked:
|
|
emp_dict["linked_agent_id"] = linked
|
|
emp_list.append(emp_dict)
|
|
result["employees"] = emp_list
|
|
except Exception:
|
|
pass
|
|
try:
|
|
result["company_profile"] = org.get_company_profile()
|
|
result["final_decider_role_id"] = org.get_final_decider_role_id(strict=False)
|
|
result["top_level_role_ids"] = org.get_top_level_role_ids()
|
|
result["org_version"] = org.current_org_version()
|
|
result["runtime_topology_version"] = org.current_runtime_topology_version()
|
|
except Exception:
|
|
pass
|
|
try:
|
|
result["runtime_policy"] = org.get_runtime_policy(org.get_company_profile())
|
|
except Exception:
|
|
pass
|
|
try:
|
|
preview_topology = org.build_runtime_delegation_topology()
|
|
result["runtime_topology_preview"] = preview_topology
|
|
if org.get_company_profile() == "custom":
|
|
policy = org.get_runtime_policy("custom")
|
|
policy_payload = policy.model_dump() if hasattr(policy, "model_dump") else dict(policy or {})
|
|
result["work_item_runtime_preview"] = build_custom_org_work_item_blueprint(
|
|
org,
|
|
runtime_topology=preview_topology,
|
|
runtime_policy=policy_payload,
|
|
).to_dict()
|
|
except Exception:
|
|
pass
|
|
|
|
store = getattr(engine, "store", None)
|
|
if store is not None:
|
|
try:
|
|
if bool(getattr(store, "is_ready", False)):
|
|
project_id = getattr(engine, "project_id", None) or "default"
|
|
if hasattr(store, "list_open_delegation_runs"):
|
|
runs = await store.list_open_delegation_runs(project_id=project_id)
|
|
else:
|
|
runs = await store.list_delegation_runs(project_id=project_id, status="running")
|
|
if runs:
|
|
active_run = runs[0]
|
|
cells = await store.list_delegation_cells(active_run.run_id)
|
|
role_sessions = await store.list_delegation_role_sessions(active_run.run_id)
|
|
work_items = await store.list_delegation_work_items(active_run.run_id)
|
|
runtime_teams = await store.list_team_instances(run_id=active_run.run_id) if hasattr(store, "list_team_instances") else []
|
|
runtime_seats = await store.list_seat_states(run_id=active_run.run_id) if hasattr(store, "list_seat_states") else []
|
|
legacy_team_payload = [
|
|
{
|
|
"cell_id": cell.cell_id,
|
|
"manager_role_id": cell.manager_role_id,
|
|
"member_role_ids": list(cell.member_role_ids),
|
|
"status": cell.status,
|
|
"is_final_decider_cell": bool((cell.metadata or {}).get("is_final_decider_cell")),
|
|
}
|
|
for cell in cells
|
|
]
|
|
legacy_seat_payload = [
|
|
{
|
|
"role_session_id": session.role_session_id,
|
|
"role_id": session.role_id,
|
|
"employee_id": session.employee_id,
|
|
"focused_work_item_id": session.focused_work_item_id,
|
|
"background_work_item_ids": list(session.background_work_item_ids),
|
|
"pending_work_item_ids": list(getattr(session, "pending_work_item_ids", []) or []),
|
|
"queue_depth": len(list(getattr(session, "pending_work_item_ids", []) or [])),
|
|
"manager_role_ids": list(session.manager_role_ids),
|
|
"status": normalize_role_runtime_status(session.status, session.focused_work_item_id),
|
|
}
|
|
for session in role_sessions
|
|
]
|
|
column_counts = {"todo": 0, "in_progress": 0, "in_review": 0, "done": 0}
|
|
blocker_count = 0
|
|
rework_count = 0
|
|
for item in work_items:
|
|
metadata = dict(item.metadata or {})
|
|
if should_hide_work_item_from_company_kanban(metadata):
|
|
continue
|
|
column = kanban_column(item.phase)
|
|
if column in column_counts:
|
|
column_counts[column] += 1
|
|
if item.blocked_reason or item.phase.value in {
|
|
"waiting_for_peer",
|
|
"waiting_for_children",
|
|
"needs_attention",
|
|
"waiting_dependencies",
|
|
}:
|
|
blocker_count += 1
|
|
if str(metadata.get("rework_feedback", "") or "").strip():
|
|
rework_count += 1
|
|
result["frontier"] = {
|
|
"run_id": active_run.run_id,
|
|
"status": active_run.status,
|
|
"lifecycle_status": getattr(active_run, "lifecycle_status", ""),
|
|
"total_cells": len(cells),
|
|
"total_role_sessions": len(role_sessions),
|
|
"total_work_items": sum(column_counts.values()),
|
|
"todo_count": column_counts["todo"],
|
|
"in_progress_count": column_counts["in_progress"],
|
|
"in_review_count": column_counts["in_review"],
|
|
"done_count": column_counts["done"],
|
|
"blocker_count": blocker_count,
|
|
"rework_count": rework_count,
|
|
"ready_count": column_counts["todo"],
|
|
"running_count": column_counts["in_progress"],
|
|
"blocked_count": blocker_count,
|
|
"waiting_count": column_counts["in_review"],
|
|
"failed_count": 0,
|
|
}
|
|
result["project_run"] = {
|
|
"run_id": active_run.run_id,
|
|
"project_id": active_run.project_id,
|
|
"session_id": active_run.session_id,
|
|
"status": active_run.status,
|
|
"lifecycle_status": getattr(active_run, "lifecycle_status", ""),
|
|
"company_profile": active_run.company_profile,
|
|
"execution_model": active_run.execution_model,
|
|
"current_revision": getattr(active_run, "current_revision", 1),
|
|
"latest_deliverable_summary": getattr(active_run, "latest_deliverable_summary", ""),
|
|
"recovery_pointer": dict(getattr(active_run, "recovery_pointer", {}) or {}),
|
|
}
|
|
dossier = dict(getattr(active_run, "project_dossier", {}) or {})
|
|
memory = getattr(engine, "memory", None)
|
|
if not dossier and memory is not None and hasattr(memory, "build_project_dossier"):
|
|
try:
|
|
dossier = await memory.build_project_dossier(
|
|
project_id=project_id,
|
|
run_id=active_run.run_id,
|
|
session_id=active_run.session_id,
|
|
)
|
|
except Exception:
|
|
dossier = {}
|
|
result["project_dossier"] = dossier
|
|
result["runtime_teams"] = [
|
|
{
|
|
"team_instance_id": team.team_instance_id,
|
|
"cell_id": team.team_id,
|
|
"team_id": team.team_id,
|
|
"manager_role_id": str((team.metadata or {}).get("lead_role_id", "") or ""),
|
|
"member_role_ids": list(team.role_ids),
|
|
"seat_ids": list(team.seat_ids),
|
|
"status": team.status,
|
|
"parent_team_id": str((team.metadata or {}).get("parent_team_id", "") or ""),
|
|
}
|
|
for team in runtime_teams
|
|
] if runtime_teams else legacy_team_payload
|
|
result["runtime_seats"] = [
|
|
{
|
|
"role_session_id": seat.role_runtime_session_id,
|
|
"role_id": seat.role_id,
|
|
"employee_id": seat.employee_id,
|
|
"team_id": seat.team_id,
|
|
"team_instance_id": seat.team_instance_id,
|
|
"seat_id": seat.seat_id,
|
|
"focused_work_item_id": seat.current_work_item_id,
|
|
"current_work_item_id": seat.current_work_item_id,
|
|
"manager_role_ids": list(seat.manager_role_ids),
|
|
"manager_seat_id": seat.manager_seat_id,
|
|
"status": normalize_role_runtime_status(seat.status, seat.current_work_item_id),
|
|
"resident_status": normalize_role_runtime_status(seat.resident_status or seat.status, seat.current_work_item_id),
|
|
"latest_notification": dict(getattr(seat, "latest_notification", {}) or {}),
|
|
"manager_digest": dict(getattr(seat, "manager_digest", {}) or {}),
|
|
}
|
|
for seat in runtime_seats
|
|
] if runtime_seats else legacy_seat_payload
|
|
result["work_items"] = [
|
|
{
|
|
"work_item_id": item.work_item_id,
|
|
"role_id": item.role_id,
|
|
"cell_id": item.cell_id,
|
|
"team_id": item.team_id,
|
|
"seat_id": item.seat_id,
|
|
"team_instance_id": item.team_instance_id,
|
|
"title": item.title,
|
|
"kind": item.kind,
|
|
"phase": item.phase.value,
|
|
"kanban_column": kanban_column(item.phase),
|
|
"batch_id": getattr(item, "batch_id", ""),
|
|
"batch_index": getattr(item, "batch_index", 0),
|
|
"deliverable_summary": clip_text(
|
|
getattr(item, "deliverable_summary", ""),
|
|
limit=1200,
|
|
marker="ui deliverable preview truncated",
|
|
).text,
|
|
"deliverable_summary_chars": len(str(getattr(item, "deliverable_summary", "") or "")),
|
|
"blocked_reason": getattr(item, "blocked_reason", ""),
|
|
"handoff_status": getattr(item, "handoff_status", ""),
|
|
"parent_work_item_id": item.parent_work_item_id,
|
|
**work_item_identity_payload(
|
|
projection_id=work_item_projection_id_from_metadata(
|
|
dict(item.metadata or {}),
|
|
fallback=str(item.projection_id or item.work_item_id or ""),
|
|
),
|
|
turn_type=work_item_turn_type_from_metadata(
|
|
dict(item.metadata or {}),
|
|
fallback=str(item.kind or ""),
|
|
),
|
|
),
|
|
"metadata": dict(item.metadata or {}),
|
|
"adaptive": dict((item.metadata or {}).get("adaptive", {}) or {}),
|
|
}
|
|
for item in work_items
|
|
]
|
|
result["seat_digests"] = [
|
|
{
|
|
"seat_id": seat.seat_id,
|
|
"team_id": seat.team_id,
|
|
"role_id": seat.role_id,
|
|
"employee_id": seat.employee_id,
|
|
"role_session_id": seat.role_runtime_session_id,
|
|
"resident_status": normalize_role_runtime_status(
|
|
seat.resident_status or seat.status,
|
|
seat.current_work_item_id,
|
|
),
|
|
"current_work_item": dict(getattr(seat, "current_work_item", {}) or {}),
|
|
"latest_notification": dict(getattr(seat, "latest_notification", {}) or {}),
|
|
"manager_digest": dict(getattr(seat, "manager_digest", {}) or {}),
|
|
}
|
|
for seat in runtime_seats
|
|
]
|
|
if hasattr(store, "get_session_links") and active_run.session_id:
|
|
links = await store.get_session_links(active_run.session_id, limit=50)
|
|
result["revision_links"] = [
|
|
{
|
|
"link_id": link.link_id,
|
|
"session_id": link.session_id,
|
|
"linked_session_id": link.linked_session_id,
|
|
"link_type": link.link_type,
|
|
"metadata": dict(link.metadata or {}),
|
|
"created_at": link.created_at.isoformat(),
|
|
}
|
|
for link in links
|
|
if str(link.link_type or "").strip() in {"continuation_of", "revision_of", "delivery_of"}
|
|
]
|
|
result["project_recovery"] = dict(getattr(active_run, "recovery_pointer", {}) or {})
|
|
except Exception:
|
|
pass
|
|
|
|
channel_mgr = getattr(engine, "channel_manager", None)
|
|
if channel_mgr:
|
|
try:
|
|
statuses = channel_mgr.get_all_statuses()
|
|
result["channels"] = [
|
|
{
|
|
"name": status.get("name", ""),
|
|
"enabled": status.get("enabled", False),
|
|
"running": status.get("running", False),
|
|
"configured": status.get("configured", False),
|
|
"available": status.get("available", False),
|
|
"ready": status.get("ready", False),
|
|
"last_error": status.get("last_error"),
|
|
"delivery_mode": status.get("delivery_mode", ""),
|
|
}
|
|
for status in statuses
|
|
]
|
|
except Exception:
|
|
pass
|
|
try:
|
|
result["installed_packages"] = [
|
|
package.model_dump() if hasattr(package, "model_dump") else package
|
|
for package in engine.config.org.installed_packages
|
|
]
|
|
except Exception:
|
|
result["installed_packages"] = []
|
|
result["runtime_teams"] = list(result.get("runtime_teams") or [])
|
|
result["runtime_seats"] = list(result.get("runtime_seats") or [])
|
|
result["work_items"] = list(result.get("work_items") or [])
|
|
result["frontier"] = dict(result.get("frontier", {}) or {})
|
|
events = self.info_events(result) if include_events else []
|
|
return ServiceResult(result, events)
|
|
|
|
def info_events(self, payload: dict[str, Any]) -> list[ServiceEvent]:
|
|
events = [ServiceEvent("org_info", payload)]
|
|
if payload.get("project_run"):
|
|
events.append(ServiceEvent("project_run_updated", payload["project_run"]))
|
|
if payload.get("seat_digests"):
|
|
events.append(ServiceEvent("seat_digest_updated", {
|
|
"run_id": dict(payload.get("project_run", {}) or {}).get("run_id"),
|
|
"seat_digests": payload["seat_digests"],
|
|
}))
|
|
if payload.get("work_items"):
|
|
events.append(ServiceEvent("work_item_batch_updated", {
|
|
"run_id": dict(payload.get("project_run", {}) or {}).get("run_id"),
|
|
"work_items": payload["work_items"],
|
|
"frontier": payload.get("frontier", {}),
|
|
}))
|
|
if payload.get("project_recovery"):
|
|
events.append(ServiceEvent("project_recovery_updated", payload["project_recovery"]))
|
|
if payload.get("revision_links"):
|
|
events.append(ServiceEvent("project_revision_created", {
|
|
"run_id": dict(payload.get("project_run", {}) or {}).get("run_id"),
|
|
"revision_links": payload["revision_links"],
|
|
}))
|
|
return events
|
|
|
|
async def export_config(self) -> ServiceResult:
|
|
from opc.core.config import build_company_org_payload_from_config
|
|
|
|
snapshot = build_org_architecture_snapshot(self.context.engine.config)
|
|
try:
|
|
config_payload = build_org_config_payload_from_config(self.context.engine.config)
|
|
except ValueError:
|
|
profile = str(getattr(self.context.engine.config.org, "company_profile", "") or "corporate").strip()
|
|
config_payload = build_company_org_payload_from_config(
|
|
self.context.engine.config,
|
|
force_profile=profile or "corporate",
|
|
)
|
|
return ServiceResult({
|
|
"config": config_payload,
|
|
"yaml": dump_org_architecture_snapshot(snapshot),
|
|
})
|
|
|
|
async def import_config(self, payload: dict[str, Any] | str, *, dry_run: bool = False) -> ServiceResult:
|
|
try:
|
|
if isinstance(payload, str):
|
|
snapshot = parse_org_architecture_snapshot(payload)
|
|
validated = apply_org_architecture_snapshot(self.context.engine.config, snapshot)
|
|
try:
|
|
validate_saved_org_id(getattr(validated.org, "organization_id", ""))
|
|
except ValueError:
|
|
if "organization_id" in snapshot:
|
|
raise
|
|
config_dir = self.context.opc_home / "config"
|
|
organization_name = str(
|
|
getattr(validated.org, "organization_name", "")
|
|
or getattr(validated.org, "company_name", "")
|
|
or "org"
|
|
).strip()
|
|
organization_id = allocate_org_config_id(config_dir, organization_name)
|
|
validated.org.organization_id = organization_id
|
|
validated.org.organization_name = organization_name
|
|
validated.org.organization_config_file = org_config_relative_path(organization_id)
|
|
else:
|
|
validated = apply_org_config_payload_to_config(self.context.engine.config, payload)
|
|
except Exception as exc:
|
|
raise ServiceError("org_config_invalid", str(exc), {"validation_errors": [str(exc)]}) from exc
|
|
before_roles = {str(getattr(role, "id", getattr(role, "role_id", "")) or "") for role in self.context.engine.config.org.roles}
|
|
after_roles = {str(getattr(role, "id", getattr(role, "role_id", "")) or "") for role in validated.org.roles}
|
|
preview = {
|
|
"roles_added": len(after_roles - before_roles),
|
|
"roles_removed": len(before_roles - after_roles),
|
|
"employees_changed": abs(len(validated.org.employees) - len(self.context.engine.config.org.employees)),
|
|
}
|
|
if dry_run:
|
|
return ServiceResult({"ok": True, "dry_run": True, "preview": preview})
|
|
self._ensure_custom_org_editable()
|
|
async with self.context.config_lock:
|
|
self.context.rebind_config(validated)
|
|
await self._persist_and_reload()
|
|
info = await self.info(include_events=True)
|
|
return ServiceResult(
|
|
{"ok": True, "dry_run": False, "preview": preview},
|
|
info.events,
|
|
)
|
|
|
|
async def saved_list(self) -> ServiceResult:
|
|
import yaml
|
|
|
|
items: list[dict[str, Any]] = []
|
|
for path in list_org_config_paths(self.context.opc_home / "config"):
|
|
try:
|
|
parsed = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
|
|
if not isinstance(parsed, dict):
|
|
continue
|
|
base_config = getattr(self.context.engine, "config", None)
|
|
if base_config is None:
|
|
from opc.core.config import OPCConfig
|
|
|
|
base_config = OPCConfig()
|
|
validated = apply_org_config_payload_to_config(
|
|
base_config,
|
|
parsed,
|
|
source_path=path,
|
|
)
|
|
org_id = str(parsed.get("organization_id") or path.stem.removeprefix("org_").removesuffix("_config"))
|
|
org_name = str(parsed.get("organization_name") or (parsed.get("company") or {}).get("name") or org_id)
|
|
items.append({
|
|
"name": org_id,
|
|
"organization_id": org_id,
|
|
"organization_name": org_name,
|
|
"filename": path.name,
|
|
"saved_at": path.stat().st_mtime,
|
|
"roles_count": len(validated.org.roles),
|
|
"employees_count": len(validated.org.employees),
|
|
})
|
|
except Exception:
|
|
continue
|
|
active_name = ""
|
|
if self.context.get_active_saved_org_name is not None:
|
|
active_name = await self.context.get_active_saved_org_name()
|
|
return ServiceResult({"orgs": items, "active_name": active_name or None})
|
|
|
|
async def saved_load(self, name: str) -> ServiceResult:
|
|
try:
|
|
organization_id = validate_saved_org_id(name)
|
|
payload, path = load_org_config_payload(self.context.opc_home / "config", organization_id)
|
|
validated = apply_org_config_payload_to_config(
|
|
self.context.engine.config,
|
|
payload,
|
|
source_path=path,
|
|
)
|
|
validate_runnable_org_config(validated, organization_id=organization_id)
|
|
except FileNotFoundError:
|
|
raise ServiceError("saved_org_not_found", "Saved organization not found", {"name": name})
|
|
except RunnableOrgConfigError as exc:
|
|
raise ServiceError("saved_org_not_runnable", str(exc), {"name": name}) from exc
|
|
except ValueError as exc:
|
|
raise ServiceError("saved_org_reserved", str(exc), {"name": name}) from exc
|
|
async with self.context.config_lock:
|
|
self.context.rebind_config(validated)
|
|
await self._persist_and_reload()
|
|
if self.context.set_active_saved_org_name is not None:
|
|
await self.context.set_active_saved_org_name(organization_id)
|
|
return ServiceResult({"ok": True, "name": organization_id, "config": payload})
|
|
|
|
async def saved_save_as(self, name: str, *, overwrite: bool = False) -> ServiceResult:
|
|
from opc.core.config import slugify_organization_name
|
|
|
|
self._ensure_custom_org_editable()
|
|
organization_name = str(name or "").strip()
|
|
if not organization_name:
|
|
raise ServiceError("organization_name_required", "organization name required")
|
|
config_dir = self.context.opc_home / "config"
|
|
preferred_id = slugify_organization_name(organization_name)
|
|
organization_id = preferred_id if overwrite else allocate_org_config_id(config_dir, organization_name, preferred_id=preferred_id)
|
|
try:
|
|
organization_id = validate_saved_org_id(organization_id)
|
|
except ValueError as exc:
|
|
raise ServiceError("saved_org_reserved", str(exc), {"name": name}) from exc
|
|
cfg = self.context.engine.config
|
|
async with self.context.config_lock:
|
|
cfg.org.organization_id = organization_id
|
|
cfg.org.organization_name = organization_name
|
|
cfg.org.organization_config_file = org_config_relative_path(organization_id)
|
|
cfg.org.company_name = organization_name
|
|
cfg.org.company_profile = "custom"
|
|
snapshot = build_org_architecture_snapshot(cfg, force_profile="custom")
|
|
snapshot["organization_id"] = organization_id
|
|
snapshot["organization_name"] = organization_name
|
|
snapshot.setdefault("company", {})["name"] = organization_name
|
|
path = write_org_config_payload(config_dir, organization_id, snapshot)
|
|
write_org_index(config_dir, organization_id)
|
|
await self._persist_and_reload()
|
|
if self.context.set_active_saved_org_name is not None:
|
|
await self.context.set_active_saved_org_name(organization_id)
|
|
return ServiceResult({
|
|
"ok": True,
|
|
"name": organization_id,
|
|
"organization_id": organization_id,
|
|
"organization_name": organization_name,
|
|
"filename": path.name,
|
|
"path": str(path),
|
|
})
|
|
|
|
async def saved_create(self, *, organization_name: str, members: list[dict[str, Any]]) -> ServiceResult:
|
|
from opc.core.config import EmployeeConfig, OPCConfig, RoleConfig, slugify_organization_name
|
|
|
|
display_name = str(organization_name or "").strip()
|
|
if not display_name:
|
|
raise ServiceError("organization_name_required", "organization name required")
|
|
|
|
if not isinstance(members, list):
|
|
raise ServiceError("org_members_required", "members must be a list")
|
|
|
|
normalized_members: list[dict[str, Any]] = []
|
|
for index, item in enumerate(members):
|
|
if not isinstance(item, dict):
|
|
continue
|
|
name = str(item.get("name") or "").strip()
|
|
if not name:
|
|
continue
|
|
normalized_members.append({
|
|
"source_index": index,
|
|
"name": name,
|
|
"responsibility": str(item.get("responsibility") or "").strip(),
|
|
"prompt": str(item.get("prompt") or "").strip(),
|
|
"reports_to_index": item.get("reports_to_index"),
|
|
})
|
|
if len(normalized_members) < 2:
|
|
raise ServiceError("org_members_required", "organization requires at least two members")
|
|
|
|
config_dir = self.context.opc_home / "config"
|
|
organization_id = allocate_org_config_id(config_dir, display_name)
|
|
role_ids: list[str] = []
|
|
used_role_ids: set[str] = set()
|
|
for idx, member in enumerate(normalized_members):
|
|
base = slugify_organization_name(member["name"], fallback=f"member_{idx + 1}")
|
|
role_id = base
|
|
suffix = 2
|
|
while role_id in used_role_ids:
|
|
tail = f"_{suffix}"
|
|
role_id = f"{base[: max(1, 64 - len(tail))].rstrip('_-') or 'member'}{tail}"
|
|
suffix += 1
|
|
used_role_ids.add(role_id)
|
|
role_ids.append(role_id)
|
|
|
|
roles: list[RoleConfig] = []
|
|
employees: list[EmployeeConfig] = []
|
|
for idx, member in enumerate(normalized_members):
|
|
raw_parent = member.get("reports_to_index")
|
|
if raw_parent in (None, "", [], {}):
|
|
reports_to = "owner" if idx == 0 else role_ids[0]
|
|
else:
|
|
try:
|
|
parent_index = int(raw_parent)
|
|
except (TypeError, ValueError) as exc:
|
|
raise ServiceError("invalid_org_member_hierarchy", "invalid reports_to_index", {"index": idx}) from exc
|
|
if parent_index < 0 or parent_index >= idx:
|
|
raise ServiceError("invalid_org_member_hierarchy", "reports_to_index must point to an earlier member", {
|
|
"index": idx,
|
|
"reports_to_index": parent_index,
|
|
})
|
|
reports_to = role_ids[parent_index]
|
|
|
|
role_id = role_ids[idx]
|
|
responsibility = member["responsibility"] or f"Owns {member['name']} responsibilities."
|
|
roles.append(RoleConfig(
|
|
id=role_id,
|
|
name=member["name"],
|
|
responsibility=responsibility,
|
|
reports_to=reports_to,
|
|
prompt_refs=[member["prompt"]] if member["prompt"] else [],
|
|
))
|
|
employees.append(EmployeeConfig(
|
|
employee_id=f"{role_id}-default-employee",
|
|
template_id="system_default_employee",
|
|
name=f"{member['name']} Default Employee",
|
|
role_id=role_id,
|
|
description=f"Default employee for the {member['name']} role.",
|
|
category="general",
|
|
metadata={
|
|
"is_default_employee": True,
|
|
"auto_created_for_role": role_id,
|
|
"employee_origin": "system_default",
|
|
"persist_to_org": True,
|
|
},
|
|
))
|
|
|
|
cfg = self.context.engine.config.model_copy(deep=True)
|
|
cfg.org.organization_id = organization_id
|
|
cfg.org.organization_name = display_name
|
|
cfg.org.organization_config_file = org_config_relative_path(organization_id)
|
|
cfg.org.company_name = display_name
|
|
cfg.org.company_profile = "custom"
|
|
cfg.org.company_profiles = ["corporate", "custom"]
|
|
cfg.org.execution_model = "actor_runtime"
|
|
cfg.org.final_decider_role_id = role_ids[0]
|
|
cfg.org.roles = roles
|
|
cfg.org.employees = employees
|
|
cfg.org.escalation_rules = []
|
|
cfg.org.runtime_policies = {}
|
|
cfg.org.talent_templates = []
|
|
cfg.org.teams = []
|
|
cfg.org.team_runtime = OPCConfig().org.team_runtime
|
|
cfg.org.installed_packages = []
|
|
validate_runnable_org_config(cfg, organization_id=organization_id)
|
|
|
|
async with self.context.config_lock:
|
|
payload = build_org_config_payload_from_config(
|
|
cfg,
|
|
organization_id=organization_id,
|
|
organization_name=display_name,
|
|
)
|
|
path = write_org_config_payload(config_dir, organization_id, payload)
|
|
write_org_index(config_dir, organization_id)
|
|
self.context.rebind_config(cfg)
|
|
org = getattr(self.context.engine, "org_engine", None)
|
|
if org and hasattr(org, "reload_from_config"):
|
|
org.reload_from_config()
|
|
if self.context.set_active_saved_org_name is not None:
|
|
await self.context.set_active_saved_org_name(organization_id)
|
|
return ServiceResult({
|
|
"ok": True,
|
|
"name": organization_id,
|
|
"organization_id": organization_id,
|
|
"organization_name": display_name,
|
|
"filename": path.name,
|
|
"path": str(path),
|
|
"roles_count": len(roles),
|
|
"employees_count": len(employees),
|
|
})
|
|
|
|
async def saved_delete(self, name: str) -> ServiceResult:
|
|
try:
|
|
organization_id = validate_saved_org_id(str(name or ""))
|
|
except ValueError as exc:
|
|
raise ServiceError("saved_org_reserved", str(exc), {"name": name}) from exc
|
|
active = ""
|
|
if self.context.get_active_saved_org_name is not None:
|
|
active = await self.context.get_active_saved_org_name()
|
|
if active == organization_id:
|
|
raise ServiceError("cannot_delete_active", "cannot_delete_active", {"name": name})
|
|
path = org_config_path(self.context.opc_home / "config", organization_id)
|
|
if path.exists():
|
|
path.unlink()
|
|
return ServiceResult({"ok": True, "name": organization_id, "organization_id": organization_id, "filename": path.name})
|
|
|
|
async def add_role(self, role_payload: dict[str, Any]) -> ServiceResult:
|
|
from opc.core.config import RoleConfig
|
|
|
|
self._ensure_custom_org_editable()
|
|
role_id = str(role_payload.get("role_id") or role_payload.get("id") or "").strip()
|
|
if not role_id:
|
|
raise ServiceError("missing_role_id", "role_id required")
|
|
cfg = self.context.engine.config.org
|
|
if any(self._role_id(role) == role_id for role in cfg.roles):
|
|
raise ServiceError("role_exists", "Role already exists", {"role_id": role_id})
|
|
role = RoleConfig(
|
|
id=role_id,
|
|
name=str(role_payload.get("name") or role_id),
|
|
responsibility=str(role_payload.get("responsibility") or role_payload.get("description") or ""),
|
|
reports_to=str(role_payload.get("reports_to") or "owner"),
|
|
icon=(str(role_payload.get("icon") or "").strip() or None),
|
|
tools=list(role_payload.get("tools", []) or []),
|
|
)
|
|
async with self.context.config_lock:
|
|
cfg.roles.append(role)
|
|
await self._persist_and_reload()
|
|
info = await self.info(include_events=True)
|
|
events = list(info.events)
|
|
if self.context.mode_state.exec_mode in {"org", "custom"}:
|
|
agents = await self._ensure_custom_role_agents()
|
|
events.append(ServiceEvent("ack", {"ok": True, "action": "agents_synced", "agents": agents}))
|
|
if self.context.broadcast_snapshot is not None:
|
|
await self.context.broadcast_snapshot()
|
|
return ServiceResult({"role": self._model_payload(role)}, events)
|
|
|
|
async def bulk_add_roles(self, roles: list[dict[str, Any]]) -> ServiceResult:
|
|
self._ensure_custom_org_editable()
|
|
added: list[str] = []
|
|
for role in roles:
|
|
role_id = str(role.get("role_id") or role.get("id") or "").strip()
|
|
if not role_id:
|
|
continue
|
|
if any(self._role_id(existing) == role_id for existing in self.context.engine.config.org.roles):
|
|
continue
|
|
result = await self.add_role(role)
|
|
added.append(str(result.payload.get("role", {}).get("id") or role_id))
|
|
if not added:
|
|
raise ServiceError("no_roles_added", "No valid roles to add")
|
|
return ServiceResult({"role_ids": added, "count": len(added)})
|
|
|
|
async def update_role(self, role_id: str, updates: dict[str, Any]) -> ServiceResult:
|
|
self._ensure_custom_org_editable()
|
|
role_id = str(role_id or "").strip()
|
|
if not role_id:
|
|
raise ServiceError("missing_role_id", "role_id required")
|
|
cfg = self.context.engine.config.org
|
|
target = next((role for role in cfg.roles if self._role_id(role) == role_id), None)
|
|
if target is None:
|
|
raise ServiceError("role_not_found", "Role not found", {"role_id": role_id})
|
|
if str(updates.get("reports_to", "") or "").strip() == role_id:
|
|
raise ServiceError("role_cycle", "Role cannot report to itself", {"role_id": role_id})
|
|
for key in ("name", "responsibility", "reports_to", "icon", "preferred_external_agent"):
|
|
if key in updates:
|
|
setattr(target, key, (str(updates[key]).strip() or None) if key in {"icon", "preferred_external_agent"} else str(updates[key]).strip())
|
|
if "reports_to" in updates:
|
|
new_reports_to = str(updates.get("reports_to") or "").strip()
|
|
if new_reports_to and new_reports_to != "owner":
|
|
role_map = {self._role_id(role): str(getattr(role, "reports_to", "") or "") for role in cfg.roles}
|
|
role_map[role_id] = new_reports_to
|
|
visited: set[str] = set()
|
|
cursor = new_reports_to
|
|
while cursor and cursor != "owner":
|
|
if cursor in visited:
|
|
raise ServiceError("role_cycle", "This would create a circular hierarchy", {"role_id": role_id})
|
|
visited.add(cursor)
|
|
cursor = role_map.get(cursor, "")
|
|
for key in ("can_spawn", "tools", "prompt_refs", "skill_refs", "capabilities"):
|
|
if key in updates:
|
|
value = updates.get(key) or []
|
|
if isinstance(value, str):
|
|
value = [item.strip() for item in value.split(",") if item.strip()]
|
|
else:
|
|
value = [str(item).strip() for item in list(value) if str(item).strip()]
|
|
setattr(target, key, list(value))
|
|
if "execution_strategy" in updates and hasattr(target, "runtime_policy"):
|
|
strategy = str(updates.get("execution_strategy") or "auto").strip()
|
|
if strategy:
|
|
target.runtime_policy.execution_strategy = strategy
|
|
async with self.context.config_lock:
|
|
await self._persist_and_reload()
|
|
info = await self.info(include_events=True)
|
|
return ServiceResult({"role": self._model_payload(target), "action": "role_updated", "role_id": role_id}, info.events)
|
|
|
|
async def delete_role(self, role_id: str) -> ServiceResult:
|
|
self._ensure_custom_org_editable()
|
|
cfg = self.context.engine.config.org
|
|
before = len(cfg.roles)
|
|
cfg.roles = [role for role in cfg.roles if self._role_id(role) != role_id]
|
|
if len(cfg.roles) == before:
|
|
raise ServiceError("role_not_found", "Role not found", {"role_id": role_id})
|
|
cfg.employees = [employee for employee in cfg.employees if getattr(employee, "role_id", "") != role_id]
|
|
for role in cfg.roles:
|
|
role.can_spawn = [item for item in list(getattr(role, "can_spawn", []) or []) if item != role_id]
|
|
if getattr(role, "reports_to", "") == role_id:
|
|
role.reports_to = "owner"
|
|
async with self.context.config_lock:
|
|
await self._persist_and_reload()
|
|
agents = await self.context.agent_store.get_all()
|
|
for agent in agents:
|
|
if agent.get("opc_role_id") == role_id:
|
|
await self.context.agent_store.remove_agent(agent["agent_id"])
|
|
if self.context.mode_state.exec_mode in {"org", "custom"}:
|
|
await self.context.agent_store.sync_custom_shadow()
|
|
info = await self.info(include_events=True)
|
|
return ServiceResult({"role_id": role_id, "action": "role_deleted"}, info.events)
|
|
|
|
async def update_runtime_policy(self, policy: dict[str, Any], *, profile: str = "custom") -> ServiceResult:
|
|
from opc.core.config import RuntimePolicyConfig
|
|
|
|
self._ensure_custom_org_editable()
|
|
current = self.context.engine.config.org.runtime_policies.get(profile)
|
|
base = current.model_dump() if hasattr(current, "model_dump") else {}
|
|
merged = self._deep_merge(base, dict(policy or {}))
|
|
self.context.engine.config.org.runtime_policies[profile] = RuntimePolicyConfig.model_validate(merged)
|
|
async with self.context.config_lock:
|
|
await self._persist_and_reload()
|
|
info = await self.info(include_events=True)
|
|
return ServiceResult(
|
|
{"profile": profile, "policy": self.context.engine.config.org.runtime_policies[profile].model_dump(), "action": "runtime_policy_updated"},
|
|
info.events,
|
|
)
|
|
|
|
async def update_org_strategy(self, *, final_decider_role_id: str | None = None) -> ServiceResult:
|
|
self._ensure_custom_org_editable()
|
|
value = str(final_decider_role_id or "").strip() or None
|
|
previous = self.context.engine.config.org.final_decider_role_id
|
|
self.context.engine.config.org.final_decider_role_id = value
|
|
org = getattr(self.context.engine, "org_engine", None)
|
|
if org and hasattr(org, "reload_from_config"):
|
|
org.reload_from_config()
|
|
validate = getattr(org, "validate_company_runtime_setup", None)
|
|
if callable(validate):
|
|
setup_error = validate()
|
|
if setup_error:
|
|
self.context.engine.config.org.final_decider_role_id = previous
|
|
org.reload_from_config()
|
|
raise ServiceError("invalid_org_strategy", str(setup_error))
|
|
async with self.context.config_lock:
|
|
await self._persist_and_reload()
|
|
info = await self.info(include_events=True)
|
|
return ServiceResult({"final_decider_role_id": value, "action": "org_strategy_updated"}, info.events)
|
|
|
|
async def reset_architecture(self) -> ServiceResult:
|
|
self._ensure_custom_org_editable()
|
|
async with self.context.config_lock:
|
|
self.context.engine.config.org.roles = []
|
|
self.context.engine.config.org.employees = []
|
|
self.context.engine.config.org.installed_packages = []
|
|
self.context.engine.config.org.runtime_policies.pop("custom", None)
|
|
await self._persist_and_reload()
|
|
try:
|
|
for agent in await self.context.agent_store.get_all():
|
|
await self.context.agent_store.remove_agent(agent["agent_id"])
|
|
await self.context.agent_store.sync_custom_shadow()
|
|
except Exception:
|
|
pass
|
|
info = await self.info(include_events=True)
|
|
return ServiceResult({"ok": True, "action": "architecture_reset"}, info.events)
|
|
|
|
async def _persist_and_reload(self) -> None:
|
|
if self.context.persist_runtime_config is not None:
|
|
self.context.persist_runtime_config()
|
|
else:
|
|
self.context.engine.config.save()
|
|
self.context.rebind_config(self.context.engine.config)
|
|
org = getattr(self.context.engine, "org_engine", None)
|
|
if org and hasattr(org, "reload_from_config"):
|
|
org.reload_from_config()
|
|
|
|
async def _ensure_custom_role_agents(self) -> list[dict[str, Any]]:
|
|
if self.context.ensure_custom_role_agents is not None:
|
|
return await self.context.ensure_custom_role_agents()
|
|
if self.context.agent_store is None:
|
|
return []
|
|
org = getattr(self.context.engine, "org_engine", None)
|
|
if self.context.mode_state.exec_mode in {"org", "custom"} and org is not None:
|
|
ensure = getattr(self.context.agent_store, "ensure_custom_role_agents", None)
|
|
if callable(ensure):
|
|
agents = await ensure(org)
|
|
if self.context.sync_role_map is not None:
|
|
await self.context.sync_role_map()
|
|
return agents
|
|
getter = getattr(self.context.agent_store, "get_all", None)
|
|
return await getter() if callable(getter) else []
|
|
|
|
@staticmethod
|
|
def _role_id(role: Any) -> str:
|
|
return str(getattr(role, "id", getattr(role, "role_id", "")) or "")
|
|
|
|
@staticmethod
|
|
def _deep_merge(base: dict[str, Any], patch: dict[str, Any]) -> dict[str, Any]:
|
|
result = dict(base)
|
|
for key, value in patch.items():
|
|
if isinstance(value, dict) and isinstance(result.get(key), dict):
|
|
result[key] = OrgService._deep_merge(result[key], value)
|
|
else:
|
|
result[key] = value
|
|
return result
|
|
|
|
@staticmethod
|
|
def _model_payload(value: Any) -> dict[str, Any]:
|
|
if hasattr(value, "model_dump"):
|
|
return value.model_dump()
|
|
if hasattr(value, "__dict__"):
|
|
return dict(value.__dict__)
|
|
return dict(value or {})
|