feat: add OPC (One Person Company) management system with Kanban board
Deploy Dashboard (Dev) / deploy-dev (push) Has been cancelled
Run Tests / Backend Tests (push) Has been cancelled
Run Tests / Frontend Tests (push) Has been cancelled
Run Tests / Test Summary (push) Has been cancelled

Phase 1 MVP implementation:
- PostgreSQL database schema for tasks, agents, executions, time tracking
- Backend API: task CRUD, agent management, automatic time tracking
- Frontend: Kanban board with drag-and-drop (Parking Lot → In Progress → Done)
- Task modal for creating/editing tasks with tags, priority, assignee
- 3 core agents seeded: PM, CTO, COO
- Agent approval workflow: CxO level requires approval, others auto-execute
- Automatic time tracking: starts on in_progress, stops on done
- RBAC integration: OPC page accessible to admin/member roles

Features:
- Drag-and-drop tasks between columns
- Assign tasks to humans or AI agents
- Priority badges (low, medium, high, urgent)
- Tags and due dates
- Task history tracking
- Time entry tracking with automatic timer
- Agent execution records with approval workflow

Next steps:
- Initialize OPC database on NAS
- Implement agent executor service
- Add WebSocket for real-time updates
- Add email notifications
- Add PDF invoice generation
This commit is contained in:
Gan, Jimmy
2026-03-31 14:31:31 +08:00
parent 51c4cd1a4e
commit 1ca019fa48
16 changed files with 1771 additions and 3 deletions
+582
View File
@@ -0,0 +1,582 @@
"""
OPC Database Module - PostgreSQL connection and schema management
"""
import asyncpg
import json
from typing import Optional, List, Dict, Any
from datetime import datetime
import os
# Database connection settings
DB_HOST = os.getenv("OPC_DB_HOST", "gitea-db")
DB_PORT = int(os.getenv("OPC_DB_PORT", "5432"))
DB_NAME = os.getenv("OPC_DB_NAME", "opc")
DB_USER = os.getenv("OPC_DB_USER", "gitea")
DB_PASSWORD = os.getenv("GITEA_DB_PASSWORD", "")
# Connection pool
_pool: Optional[asyncpg.Pool] = None
async def get_pool() -> asyncpg.Pool:
"""Get or create database connection pool"""
global _pool
if _pool is None:
_pool = await asyncpg.create_pool(
host=DB_HOST,
port=DB_PORT,
database=DB_NAME,
user=DB_USER,
password=DB_PASSWORD,
min_size=2,
max_size=10,
)
return _pool
async def close_pool():
"""Close database connection pool"""
global _pool
if _pool:
await _pool.close()
_pool = None
async def init_db():
"""Initialize database schema"""
pool = await get_pool()
async with pool.acquire() as conn:
# Create tables
await conn.execute("""
CREATE TABLE IF NOT EXISTS projects (
id SERIAL PRIMARY KEY,
name TEXT NOT NULL,
description TEXT,
status TEXT DEFAULT 'active',
client_id INTEGER,
metadata JSONB,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
await conn.execute("""
CREATE TABLE IF NOT EXISTS tasks (
id SERIAL PRIMARY KEY,
title TEXT NOT NULL,
description TEXT,
status TEXT NOT NULL DEFAULT 'parking_lot',
priority TEXT DEFAULT 'medium',
project_id INTEGER REFERENCES projects(id),
assigned_to TEXT,
assigned_type TEXT DEFAULT 'human',
tags JSONB,
metadata JSONB,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
completed_at TIMESTAMP,
due_date TIMESTAMP
)
""")
await conn.execute("""
CREATE TABLE IF NOT EXISTS task_history (
id SERIAL PRIMARY KEY,
task_id INTEGER NOT NULL REFERENCES tasks(id) ON DELETE CASCADE,
action TEXT NOT NULL,
actor TEXT NOT NULL,
actor_type TEXT DEFAULT 'human',
old_value JSONB,
new_value JSONB,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
await conn.execute("""
CREATE TABLE IF NOT EXISTS agents (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
role TEXT NOT NULL,
description TEXT,
capabilities JSONB,
system_prompt TEXT,
config JSONB,
enabled BOOLEAN DEFAULT TRUE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
await conn.execute("""
CREATE TABLE IF NOT EXISTS agent_executions (
id SERIAL PRIMARY KEY,
task_id INTEGER NOT NULL REFERENCES tasks(id) ON DELETE CASCADE,
agent_id TEXT NOT NULL REFERENCES agents(id),
status TEXT DEFAULT 'pending',
input_context JSONB,
output_result JSONB,
actions_proposed JSONB,
actions_executed JSONB,
error_message TEXT,
started_at TIMESTAMP,
completed_at TIMESTAMP,
requires_approval BOOLEAN DEFAULT FALSE,
approved_by TEXT,
approved_at TIMESTAMP
)
""")
await conn.execute("""
CREATE TABLE IF NOT EXISTS time_entries (
id SERIAL PRIMARY KEY,
task_id INTEGER REFERENCES tasks(id) ON DELETE CASCADE,
project_id INTEGER REFERENCES projects(id),
description TEXT,
duration_minutes INTEGER NOT NULL,
billable BOOLEAN DEFAULT TRUE,
hourly_rate NUMERIC(10, 2),
started_at TIMESTAMP,
ended_at TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
await conn.execute("""
CREATE TABLE IF NOT EXISTS clients (
id SERIAL PRIMARY KEY,
name TEXT NOT NULL,
email TEXT,
company TEXT,
notes TEXT,
metadata JSONB,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
# Create indexes
await conn.execute("CREATE INDEX IF NOT EXISTS idx_tasks_status ON tasks(status)")
await conn.execute("CREATE INDEX IF NOT EXISTS idx_tasks_assigned ON tasks(assigned_to, assigned_type)")
await conn.execute("CREATE INDEX IF NOT EXISTS idx_tasks_project ON tasks(project_id)")
await conn.execute("CREATE INDEX IF NOT EXISTS idx_task_history_task ON task_history(task_id)")
await conn.execute("CREATE INDEX IF NOT EXISTS idx_agent_executions_task ON agent_executions(task_id)")
await conn.execute("CREATE INDEX IF NOT EXISTS idx_agent_executions_agent ON agent_executions(agent_id)")
await conn.execute("CREATE INDEX IF NOT EXISTS idx_time_entries_task ON time_entries(task_id)")
await conn.execute("CREATE INDEX IF NOT EXISTS idx_time_entries_project ON time_entries(project_id)")
# Insert default agents
await seed_agents(conn)
async def seed_agents(conn):
"""Seed default agents"""
agents = [
{
"id": "pm",
"name": "Project Manager",
"role": "Project Manager",
"description": "Breaks down tasks, plans timelines, monitors progress",
"capabilities": ["task_breakdown", "timeline_planning", "risk_assessment", "status_reporting"],
"system_prompt": """You are a Project Manager agent. Your role is to:
- Break down large tasks into manageable subtasks
- Plan realistic timelines and identify dependencies
- Monitor project progress and identify blockers
- Provide status reports and risk assessments
When assigned a task, analyze it and propose concrete actions.""",
"config": {"model": "claude-sonnet-4-6", "temperature": 0.7, "approval_level": "auto"}
},
{
"id": "cto",
"name": "CTO",
"role": "Chief Technology Officer",
"description": "Reviews code, designs architecture, manages technical debt",
"capabilities": ["code_review", "architecture_design", "tech_debt_management", "security_audit"],
"system_prompt": """You are a CTO agent. Your role is to:
- Review code quality and architecture decisions
- Design scalable and maintainable systems
- Identify and prioritize technical debt
- Ensure security best practices
When assigned a task, provide technical leadership and propose improvements.""",
"config": {"model": "claude-sonnet-4-6", "temperature": 0.5, "approval_level": "cxo"}
},
{
"id": "coo",
"name": "COO",
"role": "Chief Operating Officer",
"description": "Optimizes processes, automates workflows, improves efficiency",
"capabilities": ["process_improvement", "workflow_automation", "efficiency_analysis"],
"system_prompt": """You are a COO agent. Your role is to:
- Identify opportunities for process improvement
- Automate repetitive workflows using n8n
- Analyze operational efficiency
- Optimize resource allocation
When assigned a task, look for automation and optimization opportunities.""",
"config": {"model": "claude-sonnet-4-6", "temperature": 0.6, "approval_level": "cxo"}
}
]
for agent in agents:
await conn.execute("""
INSERT INTO agents (id, name, role, description, capabilities, system_prompt, config, enabled)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
ON CONFLICT (id) DO UPDATE SET
name = EXCLUDED.name,
role = EXCLUDED.role,
description = EXCLUDED.description,
capabilities = EXCLUDED.capabilities,
system_prompt = EXCLUDED.system_prompt,
config = EXCLUDED.config
""", agent["id"], agent["name"], agent["role"], agent["description"],
json.dumps(agent["capabilities"]), agent["system_prompt"],
json.dumps(agent["config"]), True)
# Task operations
async def create_task(
title: str,
description: Optional[str] = None,
status: str = "parking_lot",
priority: str = "medium",
project_id: Optional[int] = None,
assigned_to: Optional[str] = None,
assigned_type: str = "human",
tags: Optional[List[str]] = None,
due_date: Optional[datetime] = None,
actor: str = "system"
) -> Dict[str, Any]:
"""Create a new task"""
pool = await get_pool()
async with pool.acquire() as conn:
row = await conn.fetchrow("""
INSERT INTO tasks (title, description, status, priority, project_id, assigned_to, assigned_type, tags, due_date)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
RETURNING id, title, description, status, priority, project_id, assigned_to, assigned_type, tags,
created_at, updated_at, completed_at, due_date
""", title, description, status, priority, project_id, assigned_to, assigned_type,
json.dumps(tags or []), due_date)
task = dict(row)
task["tags"] = json.loads(task["tags"]) if task["tags"] else []
# Log history
await conn.execute("""
INSERT INTO task_history (task_id, action, actor, actor_type, new_value)
VALUES ($1, $2, $3, $4, $5)
""", task["id"], "created", actor, "human", json.dumps(task))
return task
async def get_tasks(
status: Optional[str] = None,
assigned_to: Optional[str] = None,
project_id: Optional[int] = None,
priority: Optional[str] = None,
tags: Optional[List[str]] = None,
limit: int = 50,
offset: int = 0
) -> List[Dict[str, Any]]:
"""Get tasks with filters"""
pool = await get_pool()
async with pool.acquire() as conn:
query = "SELECT * FROM tasks WHERE 1=1"
params = []
param_count = 0
if status:
param_count += 1
query += f" AND status = ${param_count}"
params.append(status)
if assigned_to:
param_count += 1
query += f" AND assigned_to = ${param_count}"
params.append(assigned_to)
if project_id:
param_count += 1
query += f" AND project_id = ${param_count}"
params.append(project_id)
if priority:
param_count += 1
query += f" AND priority = ${param_count}"
params.append(priority)
query += " ORDER BY created_at DESC"
param_count += 1
query += f" LIMIT ${param_count}"
params.append(limit)
param_count += 1
query += f" OFFSET ${param_count}"
params.append(offset)
rows = await conn.fetch(query, *params)
tasks = [dict(row) for row in rows]
for task in tasks:
task["tags"] = json.loads(task["tags"]) if task["tags"] else []
return tasks
async def update_task(
task_id: int,
updates: Dict[str, Any],
actor: str = "system"
) -> Dict[str, Any]:
"""Update a task"""
pool = await get_pool()
async with pool.acquire() as conn:
# Get old values
old_task = await conn.fetchrow("SELECT * FROM tasks WHERE id = $1", task_id)
if not old_task:
raise ValueError(f"Task {task_id} not found")
old_values = dict(old_task)
# Build update query
set_clauses = ["updated_at = CURRENT_TIMESTAMP"]
params = []
param_count = 0
for key, value in updates.items():
if key in ["title", "description", "status", "priority", "project_id", "assigned_to", "assigned_type", "due_date", "completed_at"]:
param_count += 1
set_clauses.append(f"{key} = ${param_count}")
params.append(value)
elif key == "tags":
param_count += 1
set_clauses.append(f"tags = ${param_count}")
params.append(json.dumps(value))
param_count += 1
params.append(task_id)
query = f"UPDATE tasks SET {', '.join(set_clauses)} WHERE id = ${param_count} RETURNING *"
row = await conn.fetchrow(query, *params)
task = dict(row)
task["tags"] = json.loads(task["tags"]) if task["tags"] else []
# Log history
await conn.execute("""
INSERT INTO task_history (task_id, action, actor, actor_type, old_value, new_value)
VALUES ($1, $2, $3, $4, $5, $6)
""", task_id, "updated", actor, "human", json.dumps(old_values), json.dumps(updates))
return task
async def delete_task(task_id: int, actor: str = "system"):
"""Delete a task"""
pool = await get_pool()
async with pool.acquire() as conn:
await conn.execute("""
INSERT INTO task_history (task_id, action, actor, actor_type)
VALUES ($1, $2, $3, $4)
""", task_id, "deleted", actor, "human")
await conn.execute("DELETE FROM tasks WHERE id = $1", task_id)
# Time tracking operations
async def start_time_entry(
task_id: int,
project_id: Optional[int] = None,
actor: str = "system"
) -> Dict[str, Any]:
"""Start automatic time tracking for a task"""
pool = await get_pool()
async with pool.acquire() as conn:
# Check if there's already an active entry
active = await conn.fetchrow("""
SELECT * FROM time_entries
WHERE task_id = $1 AND ended_at IS NULL
ORDER BY started_at DESC LIMIT 1
""", task_id)
if active:
return dict(active)
# Create new time entry
row = await conn.fetchrow("""
INSERT INTO time_entries (task_id, project_id, started_at, duration_minutes)
VALUES ($1, $2, CURRENT_TIMESTAMP, 0)
RETURNING *
""", task_id, project_id)
return dict(row)
async def stop_time_entry(task_id: int) -> Optional[Dict[str, Any]]:
"""Stop automatic time tracking for a task"""
pool = await get_pool()
async with pool.acquire() as conn:
# Find active entry
active = await conn.fetchrow("""
SELECT * FROM time_entries
WHERE task_id = $1 AND ended_at IS NULL
ORDER BY started_at DESC LIMIT 1
""", task_id)
if not active:
return None
# Calculate duration and update
row = await conn.fetchrow("""
UPDATE time_entries
SET ended_at = CURRENT_TIMESTAMP,
duration_minutes = EXTRACT(EPOCH FROM (CURRENT_TIMESTAMP - started_at)) / 60
WHERE id = $1
RETURNING *
""", active["id"])
return dict(row)
async def get_time_entries(
task_id: Optional[int] = None,
project_id: Optional[int] = None
) -> List[Dict[str, Any]]:
"""Get time entries"""
pool = await get_pool()
async with pool.acquire() as conn:
query = "SELECT * FROM time_entries WHERE 1=1"
params = []
if task_id:
params.append(task_id)
query += f" AND task_id = ${len(params)}"
if project_id:
params.append(project_id)
query += f" AND project_id = ${len(params)}"
query += " ORDER BY started_at DESC"
rows = await conn.fetch(query, *params)
return [dict(row) for row in rows]
async def get_task_history(task_id: int) -> List[Dict[str, Any]]:
"""Get task history"""
pool = await get_pool()
async with pool.acquire() as conn:
rows = await conn.fetch("""
SELECT * FROM task_history
WHERE task_id = $1
ORDER BY timestamp DESC
""", task_id)
return [dict(row) for row in rows]
# Agent operations
async def get_agents(enabled_only: bool = True) -> List[Dict[str, Any]]:
"""Get all agents"""
pool = await get_pool()
async with pool.acquire() as conn:
query = "SELECT * FROM agents"
if enabled_only:
query += " WHERE enabled = TRUE"
query += " ORDER BY id"
rows = await conn.fetch(query)
agents = []
for row in rows:
agent = dict(row)
agent["capabilities"] = json.loads(agent["capabilities"]) if agent["capabilities"] else []
agent["config"] = json.loads(agent["config"]) if agent["config"] else {}
agents.append(agent)
return agents
async def get_agent(agent_id: str) -> Optional[Dict[str, Any]]:
"""Get agent by ID"""
pool = await get_pool()
async with pool.acquire() as conn:
row = await conn.fetchrow("SELECT * FROM agents WHERE id = $1", agent_id)
if not row:
return None
agent = dict(row)
agent["capabilities"] = json.loads(agent["capabilities"]) if agent["capabilities"] else []
agent["config"] = json.loads(agent["config"]) if agent["config"] else {}
return agent
async def create_agent_execution(
task_id: int,
agent_id: str,
actor: str = "system",
requires_approval: bool = False
) -> Dict[str, Any]:
"""Create agent execution record"""
pool = await get_pool()
async with pool.acquire() as conn:
# Get task and agent context
task = await conn.fetchrow("SELECT * FROM tasks WHERE id = $1", task_id)
agent = await conn.fetchrow("SELECT * FROM agents WHERE id = $1", agent_id)
if not task or not agent:
raise ValueError("Task or agent not found")
# Build input context
input_context = {
"task": dict(task),
"agent": dict(agent),
"timestamp": datetime.utcnow().isoformat()
}
# Check if agent requires approval (CxO level)
agent_config = json.loads(agent["config"]) if agent["config"] else {}
requires_approval = agent_config.get("approval_level") == "cxo"
row = await conn.fetchrow("""
INSERT INTO agent_executions (task_id, agent_id, status, input_context, requires_approval, started_at)
VALUES ($1, $2, $3, $4, $5, CURRENT_TIMESTAMP)
RETURNING *
""", task_id, agent_id, "pending", json.dumps(input_context), requires_approval)
execution = dict(row)
execution["input_context"] = json.loads(execution["input_context"]) if execution["input_context"] else {}
return execution
async def get_agent_executions(
task_id: Optional[int] = None,
agent_id: Optional[str] = None,
status: Optional[str] = None,
limit: int = 50
) -> List[Dict[str, Any]]:
"""Get agent executions"""
pool = await get_pool()
async with pool.acquire() as conn:
query = "SELECT * FROM agent_executions WHERE 1=1"
params = []
if task_id:
params.append(task_id)
query += f" AND task_id = ${len(params)}"
if agent_id:
params.append(agent_id)
query += f" AND agent_id = ${len(params)}"
if status:
params.append(status)
query += f" AND status = ${len(params)}"
query += " ORDER BY started_at DESC"
params.append(limit)
query += f" LIMIT ${len(params)}"
rows = await conn.fetch(query, *params)
executions = []
for row in rows:
execution = dict(row)
for json_field in ["input_context", "output_result", "actions_proposed", "actions_executed"]:
if execution[json_field]:
execution[json_field] = json.loads(execution[json_field])
executions.append(execution)
return executions
+15 -1
View File
@@ -6,7 +6,7 @@ from fastapi.responses import JSONResponse
from slowapi import Limiter
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from routers import docker_router, gitea, files, terminal, system, auth, chat_summary, security, passkey, totp, litellm, cc_connect, info_engine
from routers import docker_router, gitea, files, terminal, system, auth, chat_summary, security, passkey, totp, litellm, cc_connect, info_engine, opc_tasks, opc_agents
import auth as auth_module
import config
from rbac import require_page
@@ -113,6 +113,14 @@ async def monitor_containers():
@app.on_event("startup")
async def startup_event():
# Initialize OPC database
from db import opc_db
try:
await opc_db.init_db()
logging.getLogger(__name__).info("OPC database initialized")
except Exception as e:
logging.getLogger(__name__).error("Failed to initialize OPC database: %s", e)
asyncio.create_task(monitor_containers())
@app.get("/api/health")
@@ -178,6 +186,12 @@ app.include_router(info_engine.router, prefix="/api/info-engine",
app.include_router(security.router, prefix="/api/security",
dependencies=[Depends(_inject_user), Depends(require_page("security"))])
# OPC endpoints
app.include_router(opc_tasks.router, prefix="/api/opc",
dependencies=[Depends(_inject_user), Depends(require_page("opc"))])
app.include_router(opc_agents.router, prefix="/api/opc",
dependencies=[Depends(_inject_user), Depends(require_page("opc"))])
# WebSocket endpoint (auth handled inside handler via auth cookie; query token fallback temporary)
app.add_api_websocket_route("/ws/terminal", terminal.ws_endpoint)
+1 -1
View File
@@ -20,7 +20,7 @@ ROLE_GROUP_MAP = {
DEFAULT_RBAC = {
"role_defaults": {
"admin": {"pages": "*", "sidebar_links": "*"},
"member": {"pages": ["dashboard", "docker", "files", "openclaw", "chat-digest"], "sidebar_links": "*"},
"member": {"pages": ["dashboard", "docker", "files", "openclaw", "chat-digest", "opc"], "sidebar_links": "*"},
"viewer": {"pages": ["dashboard"], "sidebar_links": ["dashboard"]},
},
"user_overrides": {},
+2
View File
@@ -12,3 +12,5 @@ slowapi==0.1.9
webauthn==2.7.1
cryptography>=44.0.2
aiosqlite
asyncpg==0.29.0
reportlab==4.0.7
+10 -1
View File
@@ -7,10 +7,19 @@ from fastapi import APIRouter
from config import CC_CONNECT_URL, DOCKER_HOST
router = APIRouter()
client = docker.DockerClient(base_url=DOCKER_HOST, timeout=5)
_client = None
def get_docker_client():
"""Get or create Docker client (lazy initialization)."""
global _client
if _client is None:
_client = docker.DockerClient(base_url=DOCKER_HOST, timeout=5)
return _client
def _get_cc_connect_container():
client = get_docker_client()
matches = client.containers.list(all=True, filters={"name": "cc-connect"})
return matches[0] if matches else None
+116
View File
@@ -0,0 +1,116 @@
"""
OPC Agents Router
"""
from fastapi import APIRouter, Depends, HTTPException
from typing import Optional, List
from pydantic import BaseModel
from db import opc_db
from rbac import require_page, User, _inject_user, require_admin
router = APIRouter()
class AgentUpdate(BaseModel):
name: Optional[str] = None
description: Optional[str] = None
system_prompt: Optional[str] = None
enabled: Optional[bool] = None
class ExecutionApproval(BaseModel):
approved: bool
@router.get("/agents")
async def list_agents(
enabled_only: bool = True,
user: User = Depends(_inject_user)
):
"""List all agents"""
agents = await opc_db.get_agents(enabled_only=enabled_only)
return {"items": agents}
@router.get("/agents/{agent_id}")
async def get_agent(
agent_id: str,
user: User = Depends(_inject_user)
):
"""Get agent details"""
agent = await opc_db.get_agent(agent_id)
if not agent:
raise HTTPException(status_code=404, detail="Agent not found")
return agent
@router.put("/agents/{agent_id}", dependencies=[Depends(require_admin())])
async def update_agent(
agent_id: str,
updates: AgentUpdate,
user: User = Depends(_inject_user)
):
"""Update agent configuration (admin only)"""
# This would need an update_agent function in opc_db
raise HTTPException(status_code=501, detail="Not implemented yet")
@router.post("/agents/{agent_id}/execute")
async def trigger_agent(
agent_id: str,
task_id: int,
user: User = Depends(_inject_user)
):
"""Manually trigger agent execution"""
execution = await opc_db.create_agent_execution(
task_id=task_id,
agent_id=agent_id,
actor=user.username
)
return execution
@router.get("/executions")
async def list_executions(
task_id: Optional[int] = None,
agent_id: Optional[str] = None,
status: Optional[str] = None,
limit: int = 50,
user: User = Depends(_inject_user)
):
"""List agent executions"""
executions = await opc_db.get_agent_executions(
task_id=task_id,
agent_id=agent_id,
status=status,
limit=limit
)
return {"items": executions}
@router.get("/executions/{execution_id}")
async def get_execution(
execution_id: int,
user: User = Depends(_inject_user)
):
"""Get execution details"""
executions = await opc_db.get_agent_executions(limit=1000)
execution = next((e for e in executions if e["id"] == execution_id), None)
if not execution:
raise HTTPException(status_code=404, detail="Execution not found")
return execution
@router.post("/executions/{execution_id}/approve")
async def approve_execution(
execution_id: int,
approval: ExecutionApproval,
user: User = Depends(_inject_user)
):
"""Approve or reject agent execution (for CxO level actions)"""
# This would need an approve_execution function in opc_db
# For now, return placeholder
return {
"execution_id": execution_id,
"approved": approval.approved,
"approved_by": user.username
}
+259
View File
@@ -0,0 +1,259 @@
"""
OPC Task Management Router
"""
from fastapi import APIRouter, Depends, Query, HTTPException
from typing import Optional, List
from pydantic import BaseModel
from datetime import datetime
from db import opc_db
from rbac import require_page, User, _inject_user
router = APIRouter()
class TaskCreate(BaseModel):
title: str
description: Optional[str] = None
status: str = "parking_lot"
priority: str = "medium"
project_id: Optional[int] = None
assigned_to: Optional[str] = None
assigned_type: str = "human"
tags: Optional[List[str]] = None
due_date: Optional[datetime] = None
class TaskUpdate(BaseModel):
title: Optional[str] = None
description: Optional[str] = None
status: Optional[str] = None
priority: Optional[str] = None
project_id: Optional[int] = None
assigned_to: Optional[str] = None
assigned_type: Optional[str] = None
tags: Optional[List[str]] = None
due_date: Optional[datetime] = None
completed_at: Optional[datetime] = None
class TaskMove(BaseModel):
status: str
class TaskAssign(BaseModel):
assigned_to: str
assigned_type: str = "human"
@router.get("/tasks")
async def list_tasks(
status: Optional[str] = Query(None),
assigned_to: Optional[str] = Query(None),
project_id: Optional[int] = Query(None),
priority: Optional[str] = Query(None),
tags: Optional[str] = Query(None),
limit: int = Query(50, le=100),
offset: int = Query(0, ge=0),
user: User = Depends(_inject_user)
):
"""List tasks with filters"""
tag_list = tags.split(",") if tags else None
tasks = await opc_db.get_tasks(
status=status,
assigned_to=assigned_to,
project_id=project_id,
priority=priority,
tags=tag_list,
limit=limit,
offset=offset
)
return {"items": tasks, "total": len(tasks)}
@router.post("/tasks")
async def create_task(
task: TaskCreate,
user: User = Depends(_inject_user)
):
"""Create a new task"""
created_task = await opc_db.create_task(
title=task.title,
description=task.description,
status=task.status,
priority=task.priority,
project_id=task.project_id,
assigned_to=task.assigned_to,
assigned_type=task.assigned_type,
tags=task.tags,
due_date=task.due_date,
actor=user.username
)
# Start automatic timer if status is in_progress
if task.status == "in_progress":
await opc_db.start_time_entry(
task_id=created_task["id"],
project_id=task.project_id,
actor=user.username
)
return created_task
@router.get("/tasks/{task_id}")
async def get_task(
task_id: int,
user: User = Depends(_inject_user)
):
"""Get task details"""
tasks = await opc_db.get_tasks(limit=1, offset=0)
task = next((t for t in tasks if t["id"] == task_id), None)
if not task:
raise HTTPException(status_code=404, detail="Task not found")
return task
@router.put("/tasks/{task_id}")
async def update_task(
task_id: int,
task: TaskUpdate,
user: User = Depends(_inject_user)
):
"""Update a task"""
updates = task.dict(exclude_unset=True)
# Get current task status
tasks = await opc_db.get_tasks(limit=1000, offset=0)
current_task = next((t for t in tasks if t["id"] == task_id), None)
if not current_task:
raise HTTPException(status_code=404, detail="Task not found")
# Handle automatic time tracking
old_status = current_task["status"]
new_status = updates.get("status", old_status)
# Start timer when moving to in_progress
if old_status != "in_progress" and new_status == "in_progress":
await opc_db.start_time_entry(
task_id=task_id,
project_id=current_task.get("project_id"),
actor=user.username
)
# Stop timer when moving away from in_progress
if old_status == "in_progress" and new_status != "in_progress":
await opc_db.stop_time_entry(task_id=task_id)
# Set completed_at when moving to done
if new_status == "done" and old_status != "done":
updates["completed_at"] = datetime.utcnow()
updated_task = await opc_db.update_task(
task_id=task_id,
updates=updates,
actor=user.username
)
return updated_task
@router.delete("/tasks/{task_id}")
async def delete_task(
task_id: int,
user: User = Depends(_inject_user)
):
"""Delete a task"""
await opc_db.delete_task(task_id=task_id, actor=user.username)
return {"ok": True}
@router.put("/tasks/{task_id}/move")
async def move_task(
task_id: int,
move: TaskMove,
user: User = Depends(_inject_user)
):
"""Move task to a different column"""
updates = {"status": move.status}
# Get current task
tasks = await opc_db.get_tasks(limit=1000, offset=0)
current_task = next((t for t in tasks if t["id"] == task_id), None)
if not current_task:
raise HTTPException(status_code=404, detail="Task not found")
old_status = current_task["status"]
# Handle automatic time tracking
if old_status != "in_progress" and move.status == "in_progress":
await opc_db.start_time_entry(
task_id=task_id,
project_id=current_task.get("project_id"),
actor=user.username
)
if old_status == "in_progress" and move.status != "in_progress":
await opc_db.stop_time_entry(task_id=task_id)
if move.status == "done":
updates["completed_at"] = datetime.utcnow()
updated_task = await opc_db.update_task(
task_id=task_id,
updates=updates,
actor=user.username
)
return updated_task
@router.post("/tasks/{task_id}/assign")
async def assign_task(
task_id: int,
assign: TaskAssign,
user: User = Depends(_inject_user)
):
"""Assign task to human or agent"""
updates = {
"assigned_to": assign.assigned_to,
"assigned_type": assign.assigned_type
}
updated_task = await opc_db.update_task(
task_id=task_id,
updates=updates,
actor=user.username
)
# If assigning to agent, create execution record
if assign.assigned_type == "agent":
await opc_db.create_agent_execution(
task_id=task_id,
agent_id=assign.assigned_to,
actor=user.username
)
return updated_task
@router.get("/tasks/{task_id}/history")
async def get_task_history(
task_id: int,
user: User = Depends(_inject_user)
):
"""Get task history"""
history = await opc_db.get_task_history(task_id=task_id)
return {"items": history}
@router.get("/tasks/{task_id}/time")
async def get_task_time(
task_id: int,
user: User = Depends(_inject_user)
):
"""Get time entries for a task"""
entries = await opc_db.get_time_entries(task_id=task_id)
total_minutes = sum(e["duration_minutes"] for e in entries)
return {
"entries": entries,
"total_minutes": total_minutes,
"total_hours": round(total_minutes / 60, 2)
}