unchained/mission-control
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: simplify code formatting and improve readability across multiple files

Browse Source
This commit is contained in:
Abhimanyu Saharan
2026-02-09 20:44:05 +05:30
parent 020d02fa22
commit 8f6347dc8d
33 changed files with 393 additions and 427 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
backend/app/api/activity.py
View File

@@ -22,10 +22,7 @@ from app.models.activity_events import ActivityEvent
from app.models.agents import Agent
from app.models.boards import Board
from app.models.tasks import Task
from app.schemas.activity_events import (
ActivityEventRead,
ActivityTaskCommentFeedItemRead,
)
from app.schemas.activity_events import ActivityEventRead, ActivityTaskCommentFeedItemRead
from app.schemas.pagination import DefaultLimitOffsetPage
from app.services.organizations import (
OrganizationContext,
@@ -198,10 +195,7 @@ async def list_task_comment_feed(
def _transform(items: Sequence[Any]) -> Sequence[Any]:
rows = _coerce_task_comment_rows(items)
return [
_feed_item(event, task, board, agent)
for event, task, board, agent in rows
]
return [_feed_item(event, task, board, agent) for event, task, board, agent in rows]
return await paginate(session, statement, transformer=_transform)

87
backend/app/api/agent.py
View File

@@ -53,19 +53,9 @@ from app.schemas.gateway_coordination import (
GatewayMainAskUserResponse,
)
from app.schemas.pagination import DefaultLimitOffsetPage
from app.schemas.tasks import (
TaskCommentCreate,
TaskCommentRead,
TaskCreate,
TaskRead,
TaskUpdate,
)
from app.schemas.tasks import TaskCommentCreate, TaskCommentRead, TaskCreate, TaskRead, TaskUpdate
from app.services.activity_log import record_activity
from app.services.board_leads import (
LeadAgentOptions,
LeadAgentRequest,
ensure_board_lead_agent,
)
from app.services.board_leads import LeadAgentOptions, LeadAgentRequest, ensure_board_lead_agent
from app.services.task_dependencies import (
blocked_by_dependency_ids,
dependency_status_by_id,
@@ -212,7 +202,8 @@ async def _require_gateway_board(
board = await Board.objects.by_id(board_id).first(session)
if board is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="Board not found",
status_code=status.HTTP_404_NOT_FOUND,
detail="Board not found",
)
if board.gateway_id != gateway.id:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
@@ -322,7 +313,8 @@ async def create_task(
dependency_ids=normalized_deps,
)
blocked_by = blocked_by_dependency_ids(
dependency_ids=normalized_deps, status_by_id=dep_status,
dependency_ids=normalized_deps,
status_by_id=dep_status,
)
if blocked_by and (task.assigned_agent_id is not None or task.status != "inbox"):
@@ -393,11 +385,7 @@ async def update_task(
agent_ctx: AgentAuthContext = AGENT_CTX_DEP,
) -> TaskRead:
"""Update a task after board-level access checks."""
if (
agent_ctx.agent.board_id
and task.board_id
and agent_ctx.agent.board_id != task.board_id
):
if agent_ctx.agent.board_id and task.board_id and agent_ctx.agent.board_id != task.board_id:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
return await tasks_api.update_task(
payload=payload,
@@ -417,11 +405,7 @@ async def list_task_comments(
agent_ctx: AgentAuthContext = AGENT_CTX_DEP,
) -> LimitOffsetPage[TaskCommentRead]:
"""List comments for a task visible to the authenticated agent."""
if (
agent_ctx.agent.board_id
and task.board_id
and agent_ctx.agent.board_id != task.board_id
):
if agent_ctx.agent.board_id and task.board_id and agent_ctx.agent.board_id != task.board_id:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
return await tasks_api.list_task_comments(
task=task,
@@ -430,7 +414,8 @@ async def list_task_comments(
@router.post(
"/boards/{board_id}/tasks/{task_id}/comments", response_model=TaskCommentRead,
"/boards/{board_id}/tasks/{task_id}/comments",
response_model=TaskCommentRead,
)
async def create_task_comment(
payload: TaskCommentCreate,
@@ -439,11 +424,7 @@ async def create_task_comment(
agent_ctx: AgentAuthContext = AGENT_CTX_DEP,
) -> ActivityEvent:
"""Create a task comment on behalf of the authenticated agent."""
if (
agent_ctx.agent.board_id
and task.board_id
and agent_ctx.agent.board_id != task.board_id
):
if agent_ctx.agent.board_id and task.board_id and agent_ctx.agent.board_id != task.board_id:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
return await tasks_api.create_task_comment(
payload=payload,
@@ -454,7 +435,8 @@ async def create_task_comment(
@router.get(
"/boards/{board_id}/memory", response_model=DefaultLimitOffsetPage[BoardMemoryRead],
"/boards/{board_id}/memory",
response_model=DefaultLimitOffsetPage[BoardMemoryRead],
)
async def list_board_memory(
is_chat: bool | None = IS_CHAT_QUERY,
@@ -588,7 +570,9 @@ async def nudge_agent(
config = await _gateway_config(session, board)
try:
await ensure_session(
target.openclaw_session_id, config=config, label=target.name,
target.openclaw_session_id,
config=config,
label=target.name,
)
await send_message(
message,
@@ -605,7 +589,8 @@ async def nudge_agent(
)
await session.commit()
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc),
status_code=status.HTTP_502_BAD_GATEWAY,
detail=str(exc),
) from exc
record_activity(
session,
@@ -657,7 +642,8 @@ async def get_agent_soul(
)
except OpenClawGatewayError as exc:
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc),
status_code=status.HTTP_502_BAD_GATEWAY,
detail=str(exc),
) from exc
if isinstance(payload, str):
return payload
@@ -671,7 +657,8 @@ async def get_agent_soul(
if isinstance(nested, str):
return nested
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY, detail="Invalid gateway response",
status_code=status.HTTP_502_BAD_GATEWAY,
detail="Invalid gateway response",
)
@@ -712,7 +699,8 @@ async def update_agent_soul(
)
except OpenClawGatewayError as exc:
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc),
status_code=status.HTTP_502_BAD_GATEWAY,
detail=str(exc),
) from exc
reason = (payload.reason or "").strip()
source_url = (payload.source_url or "").strip()
@@ -770,9 +758,7 @@ async def ask_user_via_gateway_main(
correlation = payload.correlation_id.strip() if payload.correlation_id else ""
correlation_line = f"Correlation ID: {correlation}\n" if correlation else ""
preferred_channel = (payload.preferred_channel or "").strip()
channel_line = (
f"Preferred channel: {preferred_channel}\n" if preferred_channel else ""
)
channel_line = f"Preferred channel: {preferred_channel}\n" if preferred_channel else ""
tags = payload.reply_tags or ["gateway_main", "user_reply"]
tags_json = json.dumps(tags)
@@ -801,7 +787,10 @@ async def ask_user_via_gateway_main(
try:
await ensure_session(main_session_key, config=config, label="Main Agent")
await send_message(
message, session_key=main_session_key, config=config, deliver=True,
message,
session_key=main_session_key,
config=config,
deliver=True,
)
except OpenClawGatewayError as exc:
record_activity(
@@ -812,7 +801,8 @@ async def ask_user_via_gateway_main(
)
await session.commit()
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc),
status_code=status.HTTP_502_BAD_GATEWAY,
detail=str(exc),
) from exc
record_activity(
@@ -867,11 +857,7 @@ async def message_gateway_board_lead(
)
base_url = settings.base_url or "http://localhost:8000"
header = (
"GATEWAY MAIN QUESTION"
if payload.kind == "question"
else "GATEWAY MAIN HANDOFF"
)
header = "GATEWAY MAIN QUESTION" if payload.kind == "question" else "GATEWAY MAIN HANDOFF"
correlation = payload.correlation_id.strip() if payload.correlation_id else ""
correlation_line = f"Correlation ID: {correlation}\n" if correlation else ""
tags = payload.reply_tags or ["gateway_main", "lead_reply"]
@@ -903,7 +889,8 @@ async def message_gateway_board_lead(
)
await session.commit()
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc),
status_code=status.HTTP_502_BAD_GATEWAY,
detail=str(exc),
) from exc
record_activity(
@@ -946,11 +933,7 @@ async def broadcast_gateway_lead_message(
boards = list(await session.exec(statement))
base_url = settings.base_url or "http://localhost:8000"
header = (
"GATEWAY MAIN QUESTION"
if payload.kind == "question"
else "GATEWAY MAIN HANDOFF"
)
header = "GATEWAY MAIN QUESTION" if payload.kind == "question" else "GATEWAY MAIN HANDOFF"
correlation = payload.correlation_id.strip() if payload.correlation_id else ""
correlation_line = f"Correlation ID: {correlation}\n" if correlation else ""
tags = payload.reply_tags or ["gateway_main", "lead_reply"]

58
backend/app/api/agents.py
View File

@@ -23,11 +23,7 @@ from app.db import crud
from app.db.pagination import paginate
from app.db.session import async_session_maker, get_session
from app.integrations.openclaw_gateway import GatewayConfig as GatewayClientConfig
from app.integrations.openclaw_gateway import (
OpenClawGatewayError,
ensure_session,
send_message,
)
from app.integrations.openclaw_gateway import OpenClawGatewayError, ensure_session, send_message
from app.models.activity_events import ActivityEvent
from app.models.agents import Agent
from app.models.boards import Board
@@ -154,7 +150,8 @@ async def _require_board(
board = await Board.objects.by_id(board_id).first(session)
if board is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="Board not found",
status_code=status.HTTP_404_NOT_FOUND,
detail="Board not found",
)
if user is not None:
await require_board_access(session, user=user, board=board, write=write)
@@ -162,7 +159,8 @@ async def _require_board(
async def _require_gateway(
session: AsyncSession, board: Board,
session: AsyncSession,
board: Board,
) -> tuple[Gateway, GatewayClientConfig]:
if not board.gateway_id:
raise HTTPException(
@@ -246,7 +244,8 @@ def _coerce_agent_items(items: Sequence[Any]) -> list[Agent]:
async def _find_gateway_for_main_session(
session: AsyncSession, session_key: str | None,
session: AsyncSession,
session_key: str | None,
) -> Gateway | None:
if not session_key:
return None
@@ -306,7 +305,8 @@ async def _fetch_agent_events(
async def _require_user_context(
session: AsyncSession, user: User | None,
session: AsyncSession,
user: User | None,
) -> OrganizationContext:
if user is None:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
@@ -332,7 +332,8 @@ async def _require_agent_access(
if not is_org_admin(ctx.member):
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
gateway = await _find_gateway_for_main_session(
session, agent.openclaw_session_id,
session,
agent.openclaw_session_id,
)
if gateway is None or gateway.organization_id != ctx.organization.id:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
@@ -355,7 +356,10 @@ def _record_heartbeat(session: AsyncSession, agent: Agent) -> None:
def _record_instruction_failure(
session: AsyncSession, agent: Agent, error: str, action: str,
session: AsyncSession,
agent: Agent,
error: str,
action: str,
) -> None:
action_label = action.replace("_", " ").capitalize()
record_activity(
@@ -432,10 +436,7 @@ async def _ensure_unique_agent_name(
if existing_gateway:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=(
"An agent with this name already exists in this gateway "
"workspace."
),
detail=("An agent with this name already exists in this gateway " "workspace."),
)
desired_session_key = _build_session_key(requested_name)
@@ -938,7 +939,9 @@ async def _commit_heartbeat(
async def _send_wakeup_message(
agent: Agent, config: GatewayClientConfig, verb: str = "provisioned",
agent: Agent,
config: GatewayClientConfig,
verb: str = "provisioned",
) -> None:
session_key = agent.openclaw_session_id or _build_session_key(agent.name)
await ensure_session(session_key, config=config, label=agent.name)
@@ -971,7 +974,8 @@ async def list_agents(
col(Gateway.organization_id) == ctx.organization.id,
)
base_filter = or_(
base_filter, col(Agent.openclaw_session_id).in_(gateway_keys),
base_filter,
col(Agent.openclaw_session_id).in_(gateway_keys),
)
statement = select(Agent).where(base_filter)
if board_id is not None:
@@ -987,10 +991,7 @@ async def list_agents(
def _transform(items: Sequence[Any]) -> Sequence[Any]:
agents = _coerce_agent_items(items)
return [
_to_agent_read(_with_computed_status(agent), main_session_keys)
for agent in agents
]
return [_to_agent_read(_with_computed_status(agent), main_session_keys) for agent in agents]
return await paginate(session, statement, transformer=_transform)
@@ -1019,19 +1020,17 @@ async def stream_agents(
async with async_session_maker() as stream_session:
if board_id is not None:
agents = await _fetch_agent_events(
stream_session, board_id, last_seen,
stream_session,
board_id,
last_seen,
)
elif allowed_ids:
agents = await _fetch_agent_events(stream_session, None, last_seen)
agents = [
agent for agent in agents if agent.board_id in allowed_ids
]
agents = [agent for agent in agents if agent.board_id in allowed_ids]
else:
agents = []
main_session_keys = (
await _get_gateway_main_session_keys(stream_session)
if agents
else set()
await _get_gateway_main_session_keys(stream_session) if agents else set()
)
for agent in agents:
updated_at = agent.updated_at or agent.last_seen_at or utcnow()
@@ -1252,7 +1251,8 @@ async def delete_agent(
await _require_agent_access(session, agent=agent, ctx=ctx, write=True)
board = await _require_board(
session, str(agent.board_id) if agent.board_id else None,
session,
str(agent.board_id) if agent.board_id else None,
)
gateway, client_config = await _require_gateway(session, board)
try:

11
backend/app/api/approvals.py
View File

@@ -24,12 +24,7 @@ from app.core.time import utcnow
from app.db.pagination import paginate
from app.db.session import async_session_maker, get_session
from app.models.approvals import Approval
from app.schemas.approvals import (
ApprovalCreate,
ApprovalRead,
ApprovalStatus,
ApprovalUpdate,
)
from app.schemas.approvals import ApprovalCreate, ApprovalRead, ApprovalStatus, ApprovalUpdate
from app.schemas.pagination import DefaultLimitOffsetPage
if TYPE_CHECKING:
@@ -156,9 +151,7 @@ async def stream_approvals(
).one(),
)
task_ids = {
approval.task_id
for approval in approvals
if approval.task_id is not None
approval.task_id for approval in approvals if approval.task_id is not None
}
counts_by_task_id: dict[UUID, tuple[int, int]] = {}
if task_ids:

11
backend/app/api/board_group_memory.py
View File

@@ -26,21 +26,14 @@ from app.core.time import utcnow
from app.db.pagination import paginate
from app.db.session import async_session_maker, get_session
from app.integrations.openclaw_gateway import GatewayConfig as GatewayClientConfig
from app.integrations.openclaw_gateway import (
OpenClawGatewayError,
ensure_session,
send_message,
)
from app.integrations.openclaw_gateway import OpenClawGatewayError, ensure_session, send_message
from app.models.agents import Agent
from app.models.board_group_memory import BoardGroupMemory
from app.models.board_groups import BoardGroup
from app.models.boards import Board
from app.models.gateways import Gateway
from app.models.users import User
from app.schemas.board_group_memory import (
BoardGroupMemoryCreate,
BoardGroupMemoryRead,
)
from app.schemas.board_group_memory import BoardGroupMemoryCreate, BoardGroupMemoryRead
from app.schemas.pagination import DefaultLimitOffsetPage
from app.services.mentions import extract_mentions, matches_agent_mention
from app.services.organizations import (

50
backend/app/api/board_groups.py
View File

@@ -10,12 +10,7 @@ from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy import func
from sqlmodel import col, select
from app.api.deps import (
ActorContext,
require_admin_or_agent,
require_org_admin,
require_org_member,
)
from app.api.deps import ActorContext, require_admin_or_agent, require_org_admin, require_org_member
from app.core.time import utcnow
from app.db import crud
from app.db.pagination import paginate
@@ -34,10 +29,7 @@ from app.schemas.board_groups import BoardGroupCreate, BoardGroupRead, BoardGrou
from app.schemas.common import OkResponse
from app.schemas.pagination import DefaultLimitOffsetPage
from app.schemas.view_models import BoardGroupSnapshot
from app.services.agent_provisioning import (
DEFAULT_HEARTBEAT_CONFIG,
sync_gateway_agent_heartbeats,
)
from app.services.agent_provisioning import DEFAULT_HEARTBEAT_CONFIG, sync_gateway_agent_heartbeats
from app.services.board_group_snapshot import build_group_snapshot
from app.services.organizations import (
OrganizationContext,
@@ -86,8 +78,7 @@ async def _require_group_access(
return group
board_ids = [
board.id
for board in await Board.objects.filter_by(board_group_id=group_id).all(session)
board.id for board in await Board.objects.filter_by(board_group_id=group_id).all(session)
]
if not board_ids:
if is_org_admin(member):
@@ -144,7 +135,10 @@ async def get_board_group(
) -> BoardGroup:
"""Get a board group by id."""
return await _require_group_access(
session, group_id=group_id, member=ctx.member, write=False,
session,
group_id=group_id,
member=ctx.member,
write=False,
)
@@ -159,7 +153,10 @@ async def get_board_group_snapshot(
) -> BoardGroupSnapshot:
"""Get a snapshot across boards in a group."""
group = await _require_group_access(
session, group_id=group_id, member=ctx.member, write=False,
session,
group_id=group_id,
member=ctx.member,
write=False,
)
if per_board_task_limit < 0:
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY)
@@ -174,9 +171,7 @@ async def get_board_group_snapshot(
allowed_ids = set(
await list_accessible_board_ids(session, member=ctx.member, write=False),
)
snapshot.boards = [
item for item in snapshot.boards if item.board.id in allowed_ids
]
snapshot.boards = [item for item in snapshot.boards if item.board.id in allowed_ids]
return snapshot
@@ -339,14 +334,13 @@ async def update_board_group(
) -> BoardGroup:
"""Update a board group."""
group = await _require_group_access(
session, group_id=group_id, member=ctx.member, write=True,
session,
group_id=group_id,
member=ctx.member,
write=True,
)
updates = payload.model_dump(exclude_unset=True)
if (
"slug" in updates
and updates["slug"] is not None
and not updates["slug"].strip()
):
if "slug" in updates and updates["slug"] is not None and not updates["slug"].strip():
updates["slug"] = _slugify(updates.get("name") or group.name)
updates["updated_at"] = utcnow()
return await crud.patch(session, group, updates)
@@ -360,7 +354,10 @@ async def delete_board_group(
) -> OkResponse:
"""Delete a board group."""
await _require_group_access(
session, group_id=group_id, member=ctx.member, write=True,
session,
group_id=group_id,
member=ctx.member,
write=True,
)
# Boards reference groups, so clear the FK first to keep deletes simple.
@@ -378,7 +375,10 @@ async def delete_board_group(
commit=False,
)
await crud.delete_where(
session, BoardGroup, col(BoardGroup.id) == group_id, commit=False,
session,
BoardGroup,
col(BoardGroup.id) == group_id,
commit=False,
)
await session.commit()
return OkResponse()

6
backend/app/api/board_memory.py
View File

@@ -24,11 +24,7 @@ from app.core.time import utcnow
from app.db.pagination import paginate
from app.db.session import async_session_maker, get_session
from app.integrations.openclaw_gateway import GatewayConfig as GatewayClientConfig
from app.integrations.openclaw_gateway import (
OpenClawGatewayError,
ensure_session,
send_message,
)
from app.integrations.openclaw_gateway import OpenClawGatewayError, ensure_session, send_message
from app.models.agents import Agent
from app.models.board_memory import BoardMemory
from app.models.gateways import Gateway

26
backend/app/api/board_onboarding.py
View File

@@ -21,11 +21,7 @@ from app.core.config import settings
from app.core.time import utcnow
from app.db.session import get_session
from app.integrations.openclaw_gateway import GatewayConfig as GatewayClientConfig
from app.integrations.openclaw_gateway import (
OpenClawGatewayError,
ensure_session,
send_message,
)
from app.integrations.openclaw_gateway import OpenClawGatewayError, ensure_session, send_message
from app.models.board_onboarding import BoardOnboardingSession
from app.models.gateways import Gateway
from app.schemas.board_onboarding import (
@@ -39,11 +35,7 @@ from app.schemas.board_onboarding import (
BoardOnboardingUserProfile,
)
from app.schemas.boards import BoardRead
from app.services.board_leads import (
LeadAgentOptions,
LeadAgentRequest,
ensure_board_lead_agent,
)
from app.services.board_leads import LeadAgentOptions, LeadAgentRequest, ensure_board_lead_agent
if TYPE_CHECKING:
from sqlmodel.ext.asyncio.session import AsyncSession
@@ -62,7 +54,8 @@ ADMIN_AUTH_DEP = Depends(require_admin_auth)
async def _gateway_config(
session: AsyncSession, board: Board,
session: AsyncSession,
board: Board,
) -> tuple[Gateway, GatewayClientConfig]:
if not board.gateway_id:
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY)
@@ -255,11 +248,15 @@ async def start_onboarding(
try:
await ensure_session(session_key, config=config, label="Main Agent")
await send_message(
prompt, session_key=session_key, config=config, deliver=False,
prompt,
session_key=session_key,
config=config,
deliver=False,
)
except OpenClawGatewayError as exc:
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc),
status_code=status.HTTP_502_BAD_GATEWAY,
detail=str(exc),
) from exc
onboarding = BoardOnboardingSession(
@@ -311,7 +308,8 @@ async def answer_onboarding(
)
except OpenClawGatewayError as exc:
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc),
status_code=status.HTTP_502_BAD_GATEWAY,
detail=str(exc),
) from exc
onboarding.messages = messages

27
backend/app/api/boards.py
View File

@@ -104,7 +104,9 @@ async def _require_gateway_for_create(
session: AsyncSession = SESSION_DEP,
) -> Gateway:
return await _require_gateway(
session, payload.gateway_id, organization_id=ctx.organization.id,
session,
payload.gateway_id,
organization_id=ctx.organization.id,
)
@@ -155,7 +157,9 @@ async def _apply_board_update(
updates = payload.model_dump(exclude_unset=True)
if "gateway_id" in updates:
await _require_gateway(
session, updates["gateway_id"], organization_id=board.organization_id,
session,
updates["gateway_id"],
organization_id=board.organization_id,
)
if "board_group_id" in updates and updates["board_group_id"] is not None:
await _require_board_group(
@@ -164,10 +168,7 @@ async def _apply_board_update(
organization_id=board.organization_id,
)
crud.apply_updates(board, updates)
if (
updates.get("board_type") == "goal"
and (not board.objective or not board.success_metrics)
):
if updates.get("board_type") == "goal" and (not board.objective or not board.success_metrics):
# Validate only when explicitly switching to goal boards.
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
@@ -183,7 +184,8 @@ async def _apply_board_update(
async def _board_gateway(
session: AsyncSession, board: Board,
session: AsyncSession,
board: Board,
) -> tuple[Gateway | None, GatewayClientConfig | None]:
if not board.gateway_id:
return None, None
@@ -255,7 +257,8 @@ async def list_boards(
if board_group_id is not None:
statement = statement.where(col(Board.board_group_id) == board_group_id)
statement = statement.order_by(
func.lower(col(Board.name)).asc(), col(Board.created_at).desc(),
func.lower(col(Board.name)).asc(),
col(Board.created_at).desc(),
)
return await paginate(session, statement)
@@ -350,10 +353,14 @@ async def delete_board(
commit=False,
)
await crud.delete_where(
session, TaskDependency, col(TaskDependency.board_id) == board.id,
session,
TaskDependency,
col(TaskDependency.board_id) == board.id,
)
await crud.delete_where(
session, TaskFingerprint, col(TaskFingerprint.board_id) == board.id,
session,
TaskFingerprint,
col(TaskFingerprint.board_id) == board.id,
)
# Approvals can reference tasks and agents, so delete before both.

59
backend/app/api/gateway.py
View File

@@ -91,7 +91,8 @@ async def _resolve_gateway(
board = await Board.objects.by_id(params.board_id).first(session)
if board is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="Board not found",
status_code=status.HTTP_404_NOT_FOUND,
detail="Board not found",
)
if user is not None:
await require_board_access(session, user=user, board=board, write=False)
@@ -119,7 +120,10 @@ async def _resolve_gateway(
async def _require_gateway(
session: AsyncSession, board_id: str | None, *, user: User | None = None,
session: AsyncSession,
board_id: str | None,
*,
user: User | None = None,
) -> tuple[Board, GatewayClientConfig, str | None]:
params = GatewayResolveQuery(board_id=board_id)
board, config, main_session = await _resolve_gateway(
@@ -161,7 +165,9 @@ async def gateways_status(
if main_session:
try:
ensured = await ensure_session(
main_session, config=config, label="Main Agent",
main_session,
config=config,
label="Main Agent",
)
if isinstance(ensured, dict):
main_session_entry = ensured.get("entry") or ensured
@@ -178,7 +184,9 @@ async def gateways_status(
)
except OpenClawGatewayError as exc:
return GatewaysStatusResponse(
connected=False, gateway_url=config.url, error=str(exc),
connected=False,
gateway_url=config.url,
error=str(exc),
)
@@ -202,7 +210,8 @@ async def list_gateway_sessions(
sessions = await openclaw_call("sessions.list", config=config)
except OpenClawGatewayError as exc:
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc),
status_code=status.HTTP_502_BAD_GATEWAY,
detail=str(exc),
) from exc
if isinstance(sessions, dict):
sessions_list = _as_object_list(sessions.get("sessions"))
@@ -213,7 +222,9 @@ async def list_gateway_sessions(
if main_session:
try:
ensured = await ensure_session(
main_session, config=config, label="Main Agent",
main_session,
config=config,
label="Main Agent",
)
if isinstance(ensured, dict):
main_session_entry = ensured.get("entry") or ensured
@@ -233,11 +244,7 @@ async def _list_sessions(config: GatewayClientConfig) -> list[dict[str, object]]
raw_items = _as_object_list(sessions.get("sessions"))
else:
raw_items = _as_object_list(sessions)
return [
item
for item in raw_items
if isinstance(item, dict)
]
return [item for item in raw_items if isinstance(item, dict)]
async def _with_main_session(
@@ -246,9 +253,7 @@ async def _with_main_session(
config: GatewayClientConfig,
main_session: str | None,
) -> list[dict[str, object]]:
if not main_session or any(
item.get("key") == main_session for item in sessions_list
):
if not main_session or any(item.get("key") == main_session for item in sessions_list):
return sessions_list
try:
await ensure_session(main_session, config=config, label="Main Agent")
@@ -278,7 +283,8 @@ async def get_gateway_session(
sessions_list = await _list_sessions(config)
except OpenClawGatewayError as exc:
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc),
status_code=status.HTTP_502_BAD_GATEWAY,
detail=str(exc),
) from exc
sessions_list = await _with_main_session(
sessions_list,
@@ -286,12 +292,15 @@ async def get_gateway_session(
main_session=main_session,
)
session_entry = next(
(item for item in sessions_list if item.get("key") == session_id), None,
(item for item in sessions_list if item.get("key") == session_id),
None,
)
if session_entry is None and main_session and session_id == main_session:
try:
ensured = await ensure_session(
main_session, config=config, label="Main Agent",
main_session,
config=config,
label="Main Agent",
)
if isinstance(ensured, dict):
session_entry = ensured.get("entry") or ensured
@@ -299,13 +308,15 @@ async def get_gateway_session(
session_entry = None
if session_entry is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="Session not found",
status_code=status.HTTP_404_NOT_FOUND,
detail="Session not found",
)
return GatewaySessionResponse(session=session_entry)
@router.get(
"/sessions/{session_id}/history", response_model=GatewaySessionHistoryResponse,
"/sessions/{session_id}/history",
response_model=GatewaySessionHistoryResponse,
)
async def get_session_history(
session_id: str,
@@ -322,7 +333,8 @@ async def get_session_history(
history = await get_chat_history(session_id, config=config)
except OpenClawGatewayError as exc:
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc),
status_code=status.HTTP_502_BAD_GATEWAY,
detail=str(exc),
) from exc
if isinstance(history, dict) and isinstance(history.get("messages"), list):
return GatewaySessionHistoryResponse(history=history["messages"])
@@ -339,7 +351,9 @@ async def send_gateway_session_message(
) -> OkResponse:
"""Send a message into a specific gateway session."""
board, config, main_session = await _require_gateway(
session, board_id, user=auth.user,
session,
board_id,
user=auth.user,
)
if auth.user is None:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
@@ -350,7 +364,8 @@ async def send_gateway_session_message(
await send_message(payload.content, session_key=session_id, config=config)
except OpenClawGatewayError as exc:
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc),
status_code=status.HTTP_502_BAD_GATEWAY,
detail=str(exc),
) from exc
return OkResponse()

17
backend/app/api/gateways.py
View File

@@ -17,11 +17,7 @@ from app.db import crud
from app.db.pagination import paginate
from app.db.session import get_session
from app.integrations.openclaw_gateway import GatewayConfig as GatewayClientConfig
from app.integrations.openclaw_gateway import (
OpenClawGatewayError,
ensure_session,
send_message,
)
from app.integrations.openclaw_gateway import OpenClawGatewayError, ensure_session, send_message
from app.models.agents import Agent
from app.models.gateways import Gateway
from app.schemas.common import OkResponse
@@ -38,12 +34,8 @@ from app.services.agent_provisioning import (
ProvisionOptions,
provision_main_agent,
)
from app.services.template_sync import (
GatewayTemplateSyncOptions,
)
from app.services.template_sync import (
sync_gateway_templates as sync_gateway_templates_service,
)
from app.services.template_sync import GatewayTemplateSyncOptions
from app.services.template_sync import sync_gateway_templates as sync_gateway_templates_service
if TYPE_CHECKING:
from fastapi_pagination.limit_offset import LimitOffsetPage
@@ -109,7 +101,8 @@ async def _require_gateway(
)
if gateway is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="Gateway not found",
status_code=status.HTTP_404_NOT_FOUND,
detail="Gateway not found",
)
return gateway

3
backend/app/api/metrics.py
View File

@@ -8,8 +8,9 @@ from typing import Literal
from uuid import UUID
from fastapi import APIRouter, Depends, Query
from sqlalchemy import DateTime, case, func
from sqlalchemy import DateTime, case
from sqlalchemy import cast as sql_cast
from sqlalchemy import func
from sqlmodel import col, select
from sqlmodel.ext.asyncio.session import AsyncSession

76
backend/app/api/organizations.py
View File

@@ -80,7 +80,8 @@ ORG_ADMIN_DEP = Depends(require_org_admin)
def _member_to_read(
member: OrganizationMember, user: User | None,
member: OrganizationMember,
user: User | None,
) -> OrganizationMemberRead:
model = OrganizationMemberRead.model_validate(member, from_attributes=True)
if user is not None:
@@ -167,9 +168,7 @@ async def list_my_organizations(
await get_active_membership(session, auth.user)
db_user = await User.objects.by_id(auth.user.id).first(session)
active_id = (
db_user.active_organization_id if db_user else auth.user.active_organization_id
)
active_id = db_user.active_organization_id if db_user else auth.user.active_organization_id
statement = (
select(Organization, OrganizationMember)
@@ -202,7 +201,9 @@ async def set_active_org(
if auth.user is None:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
member = await set_active_organization(
session, user=auth.user, organization_id=payload.organization_id,
session,
user=auth.user,
organization_id=payload.organization_id,
)
organization = await Organization.objects.by_id(member.organization_id).first(
session,
@@ -245,7 +246,10 @@ async def delete_my_org(
group_ids = select(BoardGroup.id).where(col(BoardGroup.organization_id) == org_id)
await crud.delete_where(
session, ActivityEvent, col(ActivityEvent.task_id).in_(task_ids), commit=False,
session,
ActivityEvent,
col(ActivityEvent.task_id).in_(task_ids),
commit=False,
)
await crud.delete_where(
session,
@@ -266,10 +270,16 @@ async def delete_my_org(
commit=False,
)
await crud.delete_where(
session, Approval, col(Approval.board_id).in_(board_ids), commit=False,
session,
Approval,
col(Approval.board_id).in_(board_ids),
commit=False,
)
await crud.delete_where(
session, BoardMemory, col(BoardMemory.board_id).in_(board_ids), commit=False,
session,
BoardMemory,
col(BoardMemory.board_id).in_(board_ids),
commit=False,
)
await crud.delete_where(
session,
@@ -302,13 +312,22 @@ async def delete_my_org(
commit=False,
)
await crud.delete_where(
session, Task, col(Task.board_id).in_(board_ids), commit=False,
session,
Task,
col(Task.board_id).in_(board_ids),
commit=False,
)
await crud.delete_where(
session, Agent, col(Agent.board_id).in_(board_ids), commit=False,
session,
Agent,
col(Agent.board_id).in_(board_ids),
commit=False,
)
await crud.delete_where(
session, Board, col(Board.organization_id) == org_id, commit=False,
session,
Board,
col(Board.organization_id) == org_id,
commit=False,
)
await crud.delete_where(
session,
@@ -317,10 +336,16 @@ async def delete_my_org(
commit=False,
)
await crud.delete_where(
session, BoardGroup, col(BoardGroup.organization_id) == org_id, commit=False,
session,
BoardGroup,
col(BoardGroup.organization_id) == org_id,
commit=False,
)
await crud.delete_where(
session, Gateway, col(Gateway.organization_id) == org_id, commit=False,
session,
Gateway,
col(Gateway.organization_id) == org_id,
commit=False,
)
await crud.delete_where(
session,
@@ -342,7 +367,10 @@ async def delete_my_org(
commit=False,
)
await crud.delete_where(
session, Organization, col(Organization.id) == org_id, commit=False,
session,
Organization,
col(Organization.id) == org_id,
commit=False,
)
await session.commit()
return OkResponse()
@@ -360,14 +388,14 @@ async def get_my_membership(
).all(session)
model = _member_to_read(ctx.member, user)
model.board_access = [
OrganizationBoardAccessRead.model_validate(row, from_attributes=True)
for row in access_rows
OrganizationBoardAccessRead.model_validate(row, from_attributes=True) for row in access_rows
]
return model
@router.get(
"/me/members", response_model=DefaultLimitOffsetPage[OrganizationMemberRead],
"/me/members",
response_model=DefaultLimitOffsetPage[OrganizationMemberRead],
)
async def list_org_members(
session: AsyncSession = SESSION_DEP,
@@ -410,8 +438,7 @@ async def get_org_member(
).all(session)
model = _member_to_read(member, user)
model.board_access = [
OrganizationBoardAccessRead.model_validate(row, from_attributes=True)
for row in access_rows
OrganizationBoardAccessRead.model_validate(row, from_attributes=True) for row in access_rows
]
return model
@@ -529,9 +556,7 @@ async def remove_org_member(
user.active_organization_id = fallback_membership
else:
user.active_organization_id = (
fallback_membership.organization_id
if fallback_membership is not None
else None
fallback_membership.organization_id if fallback_membership is not None else None
)
session.add(user)
@@ -540,7 +565,8 @@ async def remove_org_member(
@router.get(
"/me/invites", response_model=DefaultLimitOffsetPage[OrganizationInviteRead],
"/me/invites",
response_model=DefaultLimitOffsetPage[OrganizationInviteRead],
)
async def list_org_invites(
session: AsyncSession = SESSION_DEP,
@@ -607,7 +633,9 @@ async def create_org_invite(
if valid_board_ids != board_ids:
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY)
await apply_invite_board_access(
session, invite=invite, entries=payload.board_access,
session,
invite=invite,
entries=payload.board_access,
)
await session.commit()
await session.refresh(invite)

97
backend/app/api/tasks.py
View File

@@ -29,11 +29,7 @@ from app.db import crud
from app.db.pagination import paginate
from app.db.session import async_session_maker, get_session
from app.integrations.openclaw_gateway import GatewayConfig as GatewayClientConfig
from app.integrations.openclaw_gateway import (
OpenClawGatewayError,
ensure_session,
send_message,
)
from app.integrations.openclaw_gateway import OpenClawGatewayError, ensure_session, send_message
from app.models.activity_events import ActivityEvent
from app.models.agents import Agent
from app.models.approvals import Approval
@@ -45,13 +41,7 @@ from app.models.tasks import Task
from app.schemas.common import OkResponse
from app.schemas.errors import BlockedTaskError
from app.schemas.pagination import DefaultLimitOffsetPage
from app.schemas.tasks import (
TaskCommentCreate,
TaskCommentRead,
TaskCreate,
TaskRead,
TaskUpdate,
)
from app.schemas.tasks import TaskCommentCreate, TaskCommentRead, TaskCreate, TaskRead, TaskUpdate
from app.services.activity_log import record_activity
from app.services.mentions import extract_mentions, matches_agent_mention
from app.services.organizations import require_board_access
@@ -263,8 +253,7 @@ async def _reconcile_dependents_for_dependency_toggle(
event_type="task.status_changed",
task_id=dependent.id,
message=(
"Task returned to inbox: dependency reopened "
f"({dependency_task.title})."
"Task returned to inbox: dependency reopened " f"({dependency_task.title})."
),
agent_id=actor_agent_id,
)
@@ -313,7 +302,8 @@ def _serialize_comment(event: ActivityEvent) -> dict[str, object]:
async def _gateway_config(
session: AsyncSession, board: Board,
session: AsyncSession,
board: Board,
) -> GatewayClientConfig | None:
if not board.gateway_id:
return None
@@ -368,10 +358,7 @@ async def _notify_agent_on_task_assign(
message = (
"TASK ASSIGNED\n"
+ "\n".join(details)
+ (
"\n\nTake action: open the task and begin work. "
"Post updates as task comments."
)
+ ("\n\nTake action: open the task and begin work. " "Post updates as task comments.")
)
try:
await _send_agent_task_message(
@@ -607,9 +594,7 @@ async def _stream_dependency_state(
rows: list[tuple[ActivityEvent, Task | None]],
) -> tuple[dict[UUID, list[UUID]], dict[UUID, str]]:
task_ids = [
task.id
for event, task in rows
if task is not None and event.event_type != "task.comment"
task.id for event, task in rows if task is not None and event.event_type != "task.comment"
]
if not task_ids:
return {}, {}
@@ -786,7 +771,8 @@ async def create_task(
dependency_ids=normalized_deps,
)
blocked_by = blocked_by_dependency_ids(
dependency_ids=normalized_deps, status_by_id=dep_status,
dependency_ids=normalized_deps,
status_by_id=dep_status,
)
if blocked_by and (task.assigned_agent_id is not None or task.status != "inbox"):
raise _blocked_task_error(blocked_by)
@@ -861,9 +847,7 @@ async def update_task(
updates = payload.model_dump(exclude_unset=True)
comment = payload.comment if "comment" in payload.model_fields_set else None
depends_on_task_ids = (
payload.depends_on_task_ids
if "depends_on_task_ids" in payload.model_fields_set
else None
payload.depends_on_task_ids if "depends_on_task_ids" in payload.model_fields_set else None
)
updates.pop("comment", None)
updates.pop("depends_on_task_ids", None)
@@ -906,13 +890,22 @@ async def delete_task(
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
await require_board_access(session, user=auth.user, board=board, write=True)
await crud.delete_where(
session, ActivityEvent, col(ActivityEvent.task_id) == task.id, commit=False,
session,
ActivityEvent,
col(ActivityEvent.task_id) == task.id,
commit=False,
)
await crud.delete_where(
session, TaskFingerprint, col(TaskFingerprint.task_id) == task.id, commit=False,
session,
TaskFingerprint,
col(TaskFingerprint.task_id) == task.id,
commit=False,
)
await crud.delete_where(
session, Approval, col(Approval.task_id) == task.id, commit=False,
session,
Approval,
col(Approval.task_id) == task.id,
commit=False,
)
await crud.delete_where(
session,
@@ -929,7 +922,8 @@ async def delete_task(
@router.get(
"/{task_id}/comments", response_model=DefaultLimitOffsetPage[TaskCommentRead],
"/{task_id}/comments",
response_model=DefaultLimitOffsetPage[TaskCommentRead],
)
async def list_task_comments(
task: Task = TASK_DEP,
@@ -1241,11 +1235,7 @@ async def _lead_apply_assignment(
status_code=status.HTTP_403_FORBIDDEN,
detail="Board leads cannot assign tasks to themselves.",
)
if (
agent.board_id
and update.task.board_id
and agent.board_id != update.task.board_id
):
if agent.board_id and update.task.board_id and agent.board_id != update.task.board_id:
raise HTTPException(status_code=status.HTTP_409_CONFLICT)
update.task.assigned_agent_id = agent.id
@@ -1256,19 +1246,13 @@ def _lead_apply_status(update: _TaskUpdateInput) -> None:
if update.task.status != "review":
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=(
"Board leads can only change status when a task is "
"in review."
),
detail=("Board leads can only change status when a task is " "in review."),
)
target_status = _required_status_value(update.updates["status"])
if target_status not in {"done", "inbox"}:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=(
"Board leads can only move review tasks to done "
"or inbox."
),
detail=("Board leads can only move review tasks to done " "or inbox."),
)
if target_status == "inbox":
update.task.assigned_agent_id = None
@@ -1397,9 +1381,7 @@ async def _apply_non_lead_agent_task_rules(
update.task.assigned_agent_id = None
update.task.in_progress_at = None
else:
update.task.assigned_agent_id = (
update.actor.agent.id if update.actor.agent else None
)
update.task.assigned_agent_id = update.actor.agent.id if update.actor.agent else None
if status_value == "in_progress":
update.task.in_progress_at = utcnow()
@@ -1462,11 +1444,7 @@ async def _apply_admin_task_rules(
agent = await Agent.objects.by_id(assigned_agent_id).first(session)
if agent is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
if (
agent.board_id
and update.task.board_id
and agent.board_id != update.task.board_id
):
if agent.board_id and update.task.board_id and agent.board_id != update.task.board_id:
raise HTTPException(status_code=status.HTTP_409_CONFLICT)
@@ -1481,9 +1459,11 @@ async def _record_task_comment_from_update(
event_type="task.comment",
message=update.comment,
task_id=update.task.id,
agent_id=update.actor.agent.id
if update.actor.actor_type == "agent" and update.actor.agent
else None,
agent_id=(
update.actor.agent.id
if update.actor.actor_type == "agent" and update.actor.agent
else None
),
)
session.add(event)
await session.commit()
@@ -1496,9 +1476,7 @@ async def _record_task_update_activity(
) -> None:
event_type, message = _task_event_details(update.task, update.previous_status)
actor_agent_id = (
update.actor.agent.id
if update.actor.actor_type == "agent" and update.actor.agent
else None
update.actor.agent.id if update.actor.actor_type == "agent" and update.actor.agent else None
)
record_activity(
session,
@@ -1525,10 +1503,7 @@ async def _notify_task_update_assignment_changes(
if (
update.task.status == "inbox"
and update.task.assigned_agent_id is None
and (
update.previous_status != "inbox"
or update.previous_assigned is not None
)
and (update.previous_status != "inbox" or update.previous_assigned is not None)
):
board = (
await Board.objects.by_id(update.task.board_id).first(session)