Files
resolutionflow/backend/app/api/endpoints/trees.py
chihlasm b3dba57bc5 feat: tenant isolation Phase 0 — app-layer filters, UUID audit, CI gate (#132)
* docs: add tenant data isolation design spec

Complete architecture plan for multi-tenant data isolation across
all layers (PostgreSQL RLS, application-layer filtering, schema
migration, testing strategy, and phased rollout checklist).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* docs: add background job isolation policy to tenant isolation spec

Documents policy for all 5 existing background jobs:
- Knowledge Flywheel and PSA Retry flagged for account_id threading
- Chat Retention already follows correct pattern (model for others)
- Maintenance Schedule Firing needs account_id in queries + Session creation
- AI Conversation Expiry approved as cross-tenant with justification

Adds approved cross-tenant query registry and Phase 2 checklist items.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* docs: add tenant isolation Phase 0 implementation plan

8 tasks covering: CRITICAL copilot hotfix, tenant_filter() helper,
get_tenant_context dependency, analytics/category/AI session gap fixes,
full UUID endpoint audit, TargetList dead code audit, teams orphan
check, and CI grep check for missing tenant filters.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* feat: add tenant_filter() helper and get_tenant_context dependency

tenant_filter(model, account_id) is the canonical app-layer tenant
scoping expression. Every query on a tenant table must use it.
build_tree_access_filter and build_step_visibility_filter updated
to call tenant_filter() internally for the account_id match.

get_tenant_context is a FastAPI dependency that returns account_id
or raises 403 if the user has no account — prevents raw access to
current_user.account_id and centralises the null check.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: scope analytics/flows/{tree_id} to requesting account

Any authenticated user could read flow analytics (session counts,
completion rates, CSAT) for any tree UUID. Now returns 404 if the
tree doesn't belong to the requesting account.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: scope category tree_count to requesting account

tree_count on GET /categories/{id} was including trees from all
accounts, leaking cross-tenant row counts.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: restrict AI session search to current user only

Search endpoint used OR(user_id, account_id), exposing other users'
problem_summary and problem_domain within the same account. Sessions
are user-scoped only — cross-user access requires explicit escalation
or sharing. List and search endpoints now behave consistently.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: add ownership check and 404 responses to ai-sessions endpoints

Cross-tenant isolation audit found:
- retry-psa-push had NO ownership check (CRITICAL) — any user could retry any session's PSA push
- save_task_lane used db.get() without ownership filter, returned 403 revealing existence
- get_session returned 403 instead of 404 for unauthorized access
- stream_documentation returned 403 instead of 404

All now use query-level user_id filtering and return 404 to avoid revealing existence.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: return 404 instead of 403 for cross-tenant session access

All session endpoints (get, update, complete, scratchpad, variables, export,
ticket-link) now return 404 instead of 403 when a user tries to access
another user's session. This prevents confirming existence of resources
across tenant boundaries.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: return 404 instead of 403 for cross-tenant tree access

get_tree and update_tree now return 404 when a user cannot access a tree
(private tree from another account). Prevents confirming resource existence
across tenant boundaries.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: return 404 instead of 403 for cross-tenant step access

get_step_or_404 now returns 404 when can_view_step or can_edit_step fails,
preventing confirmation of step existence across tenant boundaries.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: return 404 instead of 403 for cross-tenant upload access

get_upload_url and delete_upload now return 404 when the upload belongs to
a different account/user, preventing resource existence confirmation.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: return 404 instead of 403 for cross-tenant share access

revoke_share and create_share now return 404 when the caller is not the
owner, preventing resource existence confirmation across users.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: return 404 instead of 403 for cross-team tree access in maintenance schedules

_get_tree_or_403 now returns 404 when the user's team does not match,
preventing confirmation of tree existence across teams.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: return 404 instead of 403 for cross-account tag access

get_tag now returns 404 for account-specific tags that belong to another
account, preventing resource existence confirmation.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: return 404 instead of 403 for cross-account step category access

get_step_category now returns 404 for account-specific categories that
belong to another account, preventing resource existence confirmation.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* test: add cross-tenant isolation tests for Task 6 UUID audit

Tests cover:
- Tree GET/PUT returns 404 for cross-account access
- Session GET returns 404 for cross-user access
- AI session GET returns 404 for cross-user access
- AI session retry-psa-push requires ownership
- Upload URL returns 404 for cross-account access
- Share revoke returns 404 for cross-user access

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: return 404 (not 403) for get_documentation cross-user access; add missing Task 6 tests

get_documentation was revealing session existence via 403. Added pre-check
query filtering by session_id AND user_id before calling the engine.

Also add cross-tenant isolation tests for steps, tags, step_categories,
and maintenance_schedules endpoints fixed in Task 6 (TDD was skipped).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: address Task 6 quality review — rename helper, restore 403 for intra-account, add docs test

- Rename _get_tree_or_403 → _get_tree_or_404 in maintenance_schedules.py
  (function now raises 404, old name was misleading)
- Restore HTTP 403 for intra-account permission failures in update_tree:
  same-account users who can see a tree but can't edit it got 404 (wrong);
  only cross-account lookups should return 404 to avoid confirming existence
- Apply same 403/404 distinction to update_tree_visibility
- Add test: get_documentation must return 404 for cross-user session access
- Add comment documenting owner-only design for documentation endpoints

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* chore: Task 7+8 — TargetList audit, CI tenant-filter grep check

Task 7: TargetList dead code audit
- Found active code references in 12+ files across backend and frontend
  (full CRUD API + frontend page + MaintenanceScheduleSection + BatchLaunchModal)
- Decision: migrate to account_id in Phase 1 (cannot drop)
- DB row count not available from code-server — must verify from VPS SSH
  before Phase 1 migration
- Teams orphan check query documented; must run from VPS SSH before Phase 1
- Results documented in spec Section 9

Task 8: CI tenant-filter enforcement check (warn mode)
- Create backend/scripts/check_tenant_filters.py
  Scans endpoint and service files for select() on tenant tables without
  tenant_filter/account_id/user_id in surrounding context. Currently
  reports 109 warnings (Phase 1 backlog). Exits 0 (warn mode).
- Add Check tenant filter enforcement step to backend CI job
  Add --fail flag after Phase 1 backlog clears to make it blocking.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* docs: record Phase 0 audit results — 0 orphaned teams, 0 target_list rows

Both checks confirmed 2026-04-09 from production DB.
Phase 1 migration is safe to proceed.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-09 00:42:19 -04:00

1329 lines
45 KiB
Python

import logging
from datetime import datetime, timezone
from typing import Annotated, Optional
from uuid import UUID
import secrets
from fastapi import APIRouter, Depends, HTTPException, status, Query, Request
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func, or_, update
from sqlalchemy.orm import selectinload
from app.core.database import get_db
from app.models.tree import Tree
from app.models.tree_share import TreeShare
from app.models.user import User
from app.models.category import TreeCategory
from app.models.tag import TreeTag, tree_tag_assignments
from app.models.folder import UserFolder, user_folder_trees
from app.schemas.tree import (
TreeCreate, TreeUpdate, TreeResponse, TreeListResponse, CategoryInfo,
ForkCreate, ForkInfo, TreeShareCreate, TreeShareResponse,
TreeVisibilityUpdate, SharedTreeResponse, TreeValidationResponse, ValidationError,
PinnedFlowResponse, PinnedFlowsListResponse, PinnedFlowReorderRequest
)
from app.models.user_pinned_tree import UserPinnedTree
from app.api.deps import get_current_active_user, require_engineer_or_admin, require_admin, get_service_account_id
from app.core.permissions import can_edit_tree, can_access_tree
from app.core.filters import build_tree_access_filter
from app.core.subscriptions import check_tree_limit, get_account_subscription, get_plan_limits
from app.core.audit import log_audit
from app.core.config import settings
from app.core.tree_validation import can_publish_tree
from app.core.step_sync import sync_steps_from_tree, deactivate_synced_steps_for_tree
from app.services.rag_service import index_tree as rag_index_tree
router = APIRouter(prefix="/trees", tags=["trees"])
def build_tree_response(tree: Tree, author_map: dict | None = None) -> TreeListResponse:
"""Build TreeListResponse with category_info and tags."""
category_info = None
if tree.category_rel:
category_info = CategoryInfo(
id=tree.category_rel.id,
name=tree.category_rel.name,
slug=tree.category_rel.slug
)
author_name = (author_map or {}).get(tree.author_id)
return TreeListResponse(
id=tree.id,
name=tree.name,
description=tree.description,
tree_type=tree.tree_type,
category=tree.category,
category_id=tree.category_id,
category_info=category_info,
tags=tree.tag_names,
author_id=tree.author_id,
author_name=author_name,
account_id=tree.account_id,
is_active=tree.is_active,
is_public=tree.is_public,
is_default=tree.is_default,
visibility=tree.visibility,
status=tree.status,
version=tree.version,
usage_count=tree.usage_count,
created_at=tree.created_at,
updated_at=tree.updated_at
)
def build_full_tree_response(tree: Tree, parent_tree: Tree = None) -> TreeResponse:
"""Build TreeResponse with all details including category_info, tags, and fork_info."""
category_info = None
if tree.category_rel:
category_info = CategoryInfo(
id=tree.category_rel.id,
name=tree.category_rel.name,
slug=tree.category_rel.slug
)
fork_info = None
if tree.parent_tree_id or tree.fork_depth > 0:
has_updates = False
if parent_tree and tree.parent_updated_at:
has_updates = parent_tree.updated_at > tree.parent_updated_at
fork_info = ForkInfo(
parent_tree_id=tree.parent_tree_id,
root_tree_id=tree.root_tree_id,
fork_reason=tree.fork_reason,
fork_depth=tree.fork_depth,
parent_updated_at=tree.parent_updated_at,
has_parent_updates=has_updates
)
return TreeResponse(
id=tree.id,
name=tree.name,
description=tree.description,
tree_type=tree.tree_type,
category=tree.category,
category_id=tree.category_id,
category_info=category_info,
tags=tree.tag_names,
fork_info=fork_info,
tree_structure=tree.tree_structure,
intake_form=tree.intake_form,
author_id=tree.author_id,
account_id=tree.account_id,
is_active=tree.is_active,
is_public=tree.is_public,
is_default=tree.is_default,
status=tree.status,
version=tree.version,
usage_count=tree.usage_count,
created_at=tree.created_at,
updated_at=tree.updated_at,
import_metadata=tree.import_metadata
)
@router.get("", response_model=list[TreeListResponse])
async def list_trees(
db: Annotated[AsyncSession, Depends(get_db)],
current_user: Annotated[User, Depends(get_current_active_user)],
tree_type: Optional[str] = Query(None, description="Filter by tree type: troubleshooting or procedural"),
category: Optional[str] = Query(None, description="Filter by legacy category string"),
category_id: Optional[UUID] = Query(None, description="Filter by category ID"),
tags: Optional[str] = Query(None, description="Comma-separated tag slugs to filter by"),
folder_id: Optional[UUID] = Query(None, description="Filter by folder ID (user's folders only)"),
is_active: Optional[bool] = Query(None, description="Filter by active status"),
author_id: Optional[UUID] = Query(None, description="Filter by author ID"),
is_public: Optional[bool] = Query(None, description="Filter by public status"),
visibility: Optional[str] = Query(None, description="Filter by visibility: private, team, link, public"),
sort_by: Optional[str] = Query("usage_count", description="Sort order: usage_count, updated_at, created_at, name, name_desc, version"),
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=100)
):
"""List all trees with optional filters.
New filters:
- category_id: Filter by category (from tree_categories table)
- tags: Comma-separated tag slugs (e.g., "citrix,networking")
- folder_id: Show only trees in a specific folder
- sort_by: Sort order (usage_count [default], updated_at, created_at, name, name_desc, version)
"""
query = select(Tree).options(
selectinload(Tree.category_rel),
selectinload(Tree.tags)
)
# Apply filters
if tree_type:
query = query.where(Tree.tree_type == tree_type)
if category:
query = query.where(Tree.category == category)
if category_id:
query = query.where(Tree.category_id == category_id)
if is_active is not None:
query = query.where(Tree.is_active == is_active)
else:
# Default to only showing active trees
query = query.where(Tree.is_active == True)
if author_id:
query = query.where(Tree.author_id == author_id)
if is_public is not None:
query = query.where(Tree.is_public == is_public)
if visibility:
query = query.where(Tree.visibility == visibility)
# Filter by tags (all specified tags must be present)
if tags:
tag_slugs = [t.strip() for t in tags.split(",") if t.strip()]
for tag_slug in tag_slugs:
query = query.where(
Tree.tags.any(TreeTag.slug == tag_slug)
)
# Filter by folder
if folder_id:
# Verify folder belongs to user
folder_result = await db.execute(
select(UserFolder).where(
UserFolder.id == folder_id,
UserFolder.user_id == current_user.id
)
)
folder = folder_result.scalar_one_or_none()
if not folder:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Folder not found"
)
query = query.where(Tree.folders.any(UserFolder.id == folder_id))
# Apply access filter
query = query.where(build_tree_access_filter(current_user))
# Apply sorting
if sort_by == "updated_at":
query = query.order_by(Tree.updated_at.desc())
elif sort_by == "created_at":
query = query.order_by(Tree.created_at.desc())
elif sort_by == "name":
query = query.order_by(Tree.name.asc())
elif sort_by == "name_desc":
query = query.order_by(Tree.name.desc())
elif sort_by == "version":
query = query.order_by(Tree.version.desc(), Tree.updated_at.desc())
else: # Default to usage_count
query = query.order_by(Tree.usage_count.desc(), Tree.updated_at.desc())
query = query.offset(skip).limit(limit)
result = await db.execute(query)
trees = result.scalars().unique().all()
# Fetch author names in one query (avoids N+1)
author_ids = {t.author_id for t in trees if t.author_id}
author_map: dict = {}
if author_ids:
authors_result = await db.execute(
select(User.id, User.name, User.email).where(User.id.in_(author_ids))
)
for row in authors_result:
author_map[row.id] = row.name or row.email
return [build_tree_response(tree, author_map) for tree in trees]
@router.get("/categories", response_model=list[str])
async def list_categories(
db: Annotated[AsyncSession, Depends(get_db)],
current_user: Annotated[User, Depends(get_current_active_user)]
):
"""List all unique categories from trees the user can access.
Note: This returns legacy string categories. For the new category system,
use the /categories endpoint.
"""
query = select(Tree.category).where(
Tree.category.isnot(None),
Tree.is_active == True,
build_tree_access_filter(current_user)
).distinct()
result = await db.execute(query)
categories = [row[0] for row in result.all() if row[0]]
return sorted(categories)
# --- Pinned Flows Endpoints (must be before /{tree_id} to avoid route shadowing) ---
MAX_PINNED_FLOWS = 15
@router.get("/pinned", response_model=PinnedFlowsListResponse)
async def list_pinned_flows(
db: Annotated[AsyncSession, Depends(get_db)],
current_user: Annotated[User, Depends(get_current_active_user)]
):
"""List user's pinned flows, ordered by display_order."""
result = await db.execute(
select(UserPinnedTree, Tree)
.join(Tree, UserPinnedTree.tree_id == Tree.id)
.options(selectinload(Tree.category_rel))
.where(
UserPinnedTree.user_id == current_user.id,
Tree.is_active == True,
Tree.deleted_at.is_(None)
)
.order_by(UserPinnedTree.display_order, UserPinnedTree.pinned_at)
)
rows = result.all()
items = []
for pin, tree in rows:
items.append(PinnedFlowResponse(
id=pin.id,
tree_id=tree.id,
tree_name=tree.name,
tree_type=tree.tree_type,
category_emoji=None,
category_name=tree.category_rel.name if tree.category_rel else None,
pinned_at=pin.pinned_at,
display_order=pin.display_order,
))
return PinnedFlowsListResponse(items=items, count=len(items))
@router.patch("/pinned/reorder", response_model=PinnedFlowsListResponse)
async def reorder_pinned_flows(
reorder_data: PinnedFlowReorderRequest,
db: Annotated[AsyncSession, Depends(get_db)],
current_user: Annotated[User, Depends(get_current_active_user)]
):
"""Update display_order for all pinned flows."""
for item in reorder_data.order:
await db.execute(
update(UserPinnedTree)
.where(
UserPinnedTree.user_id == current_user.id,
UserPinnedTree.tree_id == item.tree_id
)
.values(display_order=item.display_order)
)
await db.commit()
# Return updated list
result = await db.execute(
select(UserPinnedTree, Tree)
.join(Tree, UserPinnedTree.tree_id == Tree.id)
.options(selectinload(Tree.category_rel))
.where(
UserPinnedTree.user_id == current_user.id,
Tree.is_active == True,
Tree.deleted_at.is_(None)
)
.order_by(UserPinnedTree.display_order, UserPinnedTree.pinned_at)
)
rows = result.all()
items = []
for pin, tree in rows:
items.append(PinnedFlowResponse(
id=pin.id,
tree_id=tree.id,
tree_name=tree.name,
tree_type=tree.tree_type,
category_emoji=None,
category_name=tree.category_rel.name if tree.category_rel else None,
pinned_at=pin.pinned_at,
display_order=pin.display_order,
))
return PinnedFlowsListResponse(items=items, count=len(items))
@router.get("/search", response_model=list[TreeListResponse])
async def search_trees(
db: Annotated[AsyncSession, Depends(get_db)],
current_user: Annotated[User, Depends(get_current_active_user)],
q: str = Query(..., min_length=2, description="Search query"),
limit: int = Query(20, ge=1, le=50)
):
"""Full-text search trees by name and description."""
# Using PostgreSQL full-text search
search_vector = func.to_tsvector('english', func.coalesce(Tree.name, '') + ' ' + func.coalesce(Tree.description, ''))
search_query = func.plainto_tsquery('english', q)
query = select(Tree).options(
selectinload(Tree.category_rel),
selectinload(Tree.tags)
).where(
Tree.is_active == True,
build_tree_access_filter(current_user),
search_vector.op('@@')(search_query)
).order_by(
func.ts_rank(search_vector, search_query).desc()
).limit(limit)
result = await db.execute(query)
trees = result.scalars().unique().all()
return [build_tree_response(tree) for tree in trees]
@router.get("/{tree_id}", response_model=TreeResponse)
async def get_tree(
tree_id: UUID,
db: Annotated[AsyncSession, Depends(get_db)],
current_user: Annotated[User, Depends(get_current_active_user)]
):
"""Get a specific tree by ID."""
result = await db.execute(
select(Tree)
.options(
selectinload(Tree.category_rel),
selectinload(Tree.tags)
)
.where(Tree.id == tree_id)
)
tree = result.scalar_one_or_none()
if not tree:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Tree not found"
)
if not tree.is_active or not can_access_tree(current_user, tree):
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Tree not found"
)
return build_full_tree_response(tree)
@router.post("", response_model=TreeResponse, status_code=status.HTTP_201_CREATED)
async def create_tree(
tree_data: TreeCreate,
db: Annotated[AsyncSession, Depends(get_db)],
current_user: Annotated[User, Depends(require_engineer_or_admin)],
service_account_id: Annotated[Optional[UUID], Depends(get_service_account_id)],
):
"""Create a new tree (engineers and admins only).
Supports:
- category_id: Assign to a category from tree_categories
- tags: List of tag names to assign (creates new tags if needed)
- status: draft or published (published requires validation)
"""
# Validate tree if status is 'published'
if tree_data.status == 'published':
# Convert intake_form to dicts for validation
intake_form_dicts = None
if tree_data.intake_form:
intake_form_dicts = [f.model_dump() for f in tree_data.intake_form]
can_publish, validation_errors = can_publish_tree(
tree_data.tree_structure,
tree_data.name,
tree_data.description,
tree_type=tree_data.tree_type,
intake_form=intake_form_dicts,
)
if not can_publish:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail={
"message": "Cannot publish tree with validation errors",
"errors": validation_errors
}
)
# Only admins can create default/system trees
is_default = tree_data.is_default and current_user.is_super_admin
# Verify category exists if provided
if tree_data.category_id:
cat_result = await db.execute(
select(TreeCategory).where(TreeCategory.id == tree_data.category_id)
)
category = cat_result.scalar_one_or_none()
if not category:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Category not found"
)
# Check category access
if category.account_id and category.account_id != current_user.account_id and not current_user.is_super_admin:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You don't have access to this category"
)
# Convert intake_form Pydantic models to dicts for JSONB storage
intake_form_data = None
if tree_data.intake_form:
intake_form_data = [f.model_dump(exclude_none=True) for f in tree_data.intake_form]
new_tree = Tree(
name=tree_data.name,
description=tree_data.description,
category=tree_data.category,
category_id=tree_data.category_id,
tree_type=tree_data.tree_type,
tree_structure=tree_data.tree_structure,
intake_form=intake_form_data,
author_id=service_account_id if is_default else current_user.id,
account_id=None if is_default else current_user.account_id,
is_public=True if is_default else tree_data.is_public, # Default trees are always public
is_default=is_default,
status=tree_data.status
)
# Check subscription tree limit
if not is_default and current_user.account_id:
can_create, limit, count = await check_tree_limit(current_user.account_id, db)
if not can_create:
raise HTTPException(
status_code=status.HTTP_402_PAYMENT_REQUIRED,
detail=f"Tree limit reached ({count}/{limit}). Upgrade your plan to create more trees."
)
db.add(new_tree)
await db.flush() # Get the ID
# Re-check tree limit after flush to close the TOCTOU race window:
# two concurrent creates could both pass the pre-check, but only one
# should succeed when the limit is exactly reached.
if not is_default and current_user.account_id:
post_count = await db.scalar(
select(func.count(Tree.id)).where(
Tree.account_id == current_user.account_id,
Tree.deleted_at.is_(None),
)
)
sub = await get_account_subscription(current_user.account_id, db)
if sub:
limits = await get_plan_limits(sub.plan, db)
if limits and limits.max_trees and (post_count or 0) > limits.max_trees:
await db.rollback()
raise HTTPException(
status_code=status.HTTP_402_PAYMENT_REQUIRED,
detail=f"Tree limit reached ({limits.max_trees}/{limits.max_trees}). Upgrade your plan to create more trees."
)
# Handle tags
if tree_data.tags:
tree_account_id = new_tree.account_id or (current_user.account_id if not current_user.is_super_admin else None)
# Collect tags to add
tags_to_add = []
for tag_name in tree_data.tags:
slug = TreeTag.slugify(tag_name)
# Try to find existing tag
tag_query = select(TreeTag).where(
TreeTag.slug == slug,
or_(
TreeTag.account_id.is_(None),
TreeTag.account_id == tree_account_id
)
)
tag_result = await db.execute(tag_query)
tag = tag_result.scalar_one_or_none()
if not tag:
# Create new tag
tag = TreeTag(
name=tag_name,
slug=slug,
account_id=tree_account_id,
created_by=current_user.id
)
db.add(tag)
await db.flush()
tags_to_add.append(tag)
# Use direct SQL insert for the junction table to avoid lazy load issues
from app.models.tag import tree_tag_assignments
for tag in tags_to_add:
await db.execute(
tree_tag_assignments.insert().values(
tree_id=new_tree.id,
tag_id=tag.id,
assigned_by=current_user.id
)
)
# Atomically increment (SQL-level to avoid lost updates from concurrent requests)
await db.execute(
update(TreeTag).where(TreeTag.id == tag.id).values(usage_count=TreeTag.usage_count + 1)
)
await db.commit()
# Reload with relationships
result = await db.execute(
select(Tree)
.options(
selectinload(Tree.category_rel),
selectinload(Tree.tags)
)
.where(Tree.id == new_tree.id)
)
tree = result.scalar_one()
# Index tree for RAG (best-effort, don't fail the request)
try:
await rag_index_tree(tree.id, db)
await db.commit()
except Exception:
logging.getLogger(__name__).warning("RAG indexing failed for tree %s", tree.id)
return build_full_tree_response(tree)
@router.put("/{tree_id}", response_model=TreeResponse)
async def update_tree(
tree_id: UUID,
tree_data: TreeUpdate,
db: Annotated[AsyncSession, Depends(get_db)],
current_user: Annotated[User, Depends(require_engineer_or_admin)],
service_account_id: Annotated[Optional[UUID], Depends(get_service_account_id)],
):
"""Update an existing tree (engineers and admins only).
Supports:
- category_id: Change category assignment
- tags: Replace all tags on the tree
- status: Update status (requires validation when publishing)
"""
result = await db.execute(
select(Tree)
.options(
selectinload(Tree.category_rel),
selectinload(Tree.tags)
)
.where(Tree.id == tree_id)
)
tree = result.scalar_one_or_none()
if not tree:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Tree not found"
)
if not can_edit_tree(current_user, tree):
# If the user can see this tree (same account, team visibility), give a 403 with
# a clear message — returning 404 here would be confusing since GET returns 200.
# For truly inaccessible trees (cross-account), return 404 to avoid confirming existence.
if can_access_tree(current_user, tree):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You do not have permission to edit this flow"
)
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Tree not found"
)
# Extract tags for separate handling
update_data = tree_data.model_dump(exclude_unset=True)
tags_data = update_data.pop("tags", None)
# Validate if transitioning to published status
if "status" in update_data and update_data["status"] == 'published':
# Get the final tree structure and name after update
final_tree_structure = update_data.get("tree_structure", tree.tree_structure)
final_name = update_data.get("name", tree.name)
final_description = update_data.get("description", tree.description)
final_tree_type = update_data.get("tree_type", tree.tree_type)
final_intake_form = update_data.get("intake_form", tree.intake_form)
can_publish, validation_errors = can_publish_tree(
final_tree_structure,
final_name,
final_description,
tree_type=final_tree_type,
intake_form=final_intake_form,
)
if not can_publish:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail={
"message": "Cannot publish tree with validation errors",
"errors": validation_errors
}
)
# Verify new category if provided
if "category_id" in update_data and update_data["category_id"]:
cat_result = await db.execute(
select(TreeCategory).where(TreeCategory.id == update_data["category_id"])
)
category = cat_result.scalar_one_or_none()
if not category:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Category not found"
)
if category.account_id and category.account_id != current_user.account_id and not current_user.is_super_admin:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You don't have access to this category"
)
# Update basic fields
for field, value in update_data.items():
setattr(tree, field, value)
# Keep visibility and is_public in sync
if tree_data.is_public is not None:
if tree_data.is_public and tree.visibility not in ('public',):
tree.visibility = 'public'
elif not tree_data.is_public and tree.visibility == 'public':
tree.visibility = 'team' # downgrade from public to team
# Increment version if tree structure changed
if "tree_structure" in update_data:
tree.version += 1
# Sync steps to step library on publish transition only
if update_data.get("status") == 'published':
_structure = update_data.get("tree_structure", tree.tree_structure)
_type = update_data.get("tree_type", tree.tree_type)
_is_public = update_data.get("is_public", tree.is_public)
await sync_steps_from_tree(
db=db,
tree_id=tree.id,
tree_type=_type,
tree_structure=_structure,
author_id=tree.author_id,
account_id=tree.account_id,
is_public=_is_public,
service_account_id=service_account_id,
)
# Handle tags replacement
if tags_data is not None:
from app.models.tag import tree_tag_assignments
# Atomically decrement usage count for old tags
old_tag_ids = [tag.id for tag in tree.tags]
if old_tag_ids:
await db.execute(
update(TreeTag)
.where(TreeTag.id.in_(old_tag_ids))
.values(usage_count=func.greatest(TreeTag.usage_count - 1, 0))
)
# Delete existing tag assignments using direct SQL
await db.execute(
tree_tag_assignments.delete().where(
tree_tag_assignments.c.tree_id == tree.id
)
)
# Add new tags
tree_account_id = tree.account_id or (current_user.account_id if not current_user.is_super_admin else None)
added_tag_ids = set()
for tag_name in tags_data:
slug = TreeTag.slugify(tag_name)
tag_query = select(TreeTag).where(
TreeTag.slug == slug,
or_(
TreeTag.account_id.is_(None),
TreeTag.account_id == tree_account_id
)
)
tag_result = await db.execute(tag_query)
tag = tag_result.scalar_one_or_none()
if not tag:
tag = TreeTag(
name=tag_name,
slug=slug,
account_id=tree_account_id,
created_by=current_user.id
)
db.add(tag)
await db.flush()
if tag.id not in added_tag_ids:
await db.execute(
tree_tag_assignments.insert().values(
tree_id=tree.id,
tag_id=tag.id,
assigned_by=current_user.id
)
)
added_tag_ids.add(tag.id)
# Atomically increment (SQL-level to avoid lost updates)
await db.execute(
update(TreeTag).where(TreeTag.id == tag.id).values(usage_count=TreeTag.usage_count + 1)
)
await db.commit()
# Reload with relationships
result = await db.execute(
select(Tree)
.options(
selectinload(Tree.category_rel),
selectinload(Tree.tags)
)
.where(Tree.id == tree_id)
)
tree = result.scalar_one()
# Re-index tree for RAG (best-effort)
try:
await rag_index_tree(tree.id, db)
await db.commit()
except Exception:
logging.getLogger(__name__).warning("RAG re-indexing failed for tree %s", tree.id)
return build_full_tree_response(tree)
@router.delete("/{tree_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_tree(
tree_id: UUID,
db: Annotated[AsyncSession, Depends(get_db)],
current_user: Annotated[User, Depends(require_admin)]
):
"""Soft delete a tree (admin only). Sets deleted_at timestamp and is_active=False."""
result = await db.execute(select(Tree).where(Tree.id == tree_id))
tree = result.scalar_one_or_none()
if not tree:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Tree not found"
)
tree.is_active = False
tree.deleted_at = datetime.now(timezone.utc)
tree.deleted_by = current_user.id
# Clean up folder assignments
await db.execute(
user_folder_trees.delete().where(user_folder_trees.c.tree_id == tree.id)
)
# Decrement usage_count on associated tags (floor at 0)
tag_ids_result = await db.execute(
select(tree_tag_assignments.c.tag_id).where(
tree_tag_assignments.c.tree_id == tree.id
)
)
tag_ids = [row[0] for row in tag_ids_result.fetchall()]
if tag_ids:
await db.execute(
update(TreeTag)
.where(TreeTag.id.in_(tag_ids))
.values(usage_count=func.greatest(TreeTag.usage_count - 1, 0))
)
# Clean up tag assignments
await db.execute(
tree_tag_assignments.delete().where(tree_tag_assignments.c.tree_id == tree.id)
)
# Deactivate any synced step library entries before deletion
# (must happen before db.delete/commit — FK SET NULL would lose the reference)
await deactivate_synced_steps_for_tree(db, tree.id)
await log_audit(db, current_user.id, "tree.delete", "tree", tree.id,
{"tree_name": tree.name})
await db.commit()
return None
# --- Fork Endpoints ---
@router.post("/{tree_id}/fork", response_model=TreeResponse, status_code=status.HTTP_201_CREATED)
async def fork_tree(
tree_id: UUID,
fork_data: ForkCreate,
db: Annotated[AsyncSession, Depends(get_db)],
current_user: Annotated[User, Depends(require_engineer_or_admin)]
):
"""Fork a tree to create a personal copy.
Engineers can fork any tree they can access (public, account, or default).
Fork inherits tree_structure but gets new ownership.
"""
# Load parent tree
result = await db.execute(
select(Tree)
.options(selectinload(Tree.category_rel), selectinload(Tree.tags))
.where(Tree.id == tree_id)
)
parent = result.scalar_one_or_none()
if not parent:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Tree not found"
)
if not can_access_tree(current_user, parent):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You don't have access to this tree"
)
# Check subscription tree limit
if current_user.account_id:
can_create, limit, count = await check_tree_limit(current_user.account_id, db)
if not can_create:
raise HTTPException(
status_code=status.HTTP_402_PAYMENT_REQUIRED,
detail=f"Tree limit reached ({count}/{limit}). Upgrade your plan to create more trees."
)
# Build fork
fork_name = fork_data.name or f"Fork of {parent.name}"
fork = Tree(
name=fork_name,
description=parent.description,
category=parent.category,
category_id=parent.category_id,
tree_type=parent.tree_type,
tree_structure=parent.tree_structure,
intake_form=parent.intake_form,
author_id=current_user.id,
account_id=current_user.account_id,
is_public=False,
is_default=False,
version=1,
# Fork tracking
parent_tree_id=parent.id,
fork_reason=fork_data.fork_reason,
parent_updated_at=parent.updated_at,
# Lineage tracking
root_tree_id=parent.root_tree_id if parent.root_tree_id else parent.id,
fork_depth=parent.fork_depth + 1,
)
db.add(fork)
await db.flush()
await log_audit(db, current_user.id, "tree.fork", "tree", fork.id,
{"parent_tree_id": str(parent.id), "parent_name": parent.name,
"fork_reason": fork_data.fork_reason})
await db.commit()
# Reload with relationships
result = await db.execute(
select(Tree)
.options(selectinload(Tree.category_rel), selectinload(Tree.tags))
.where(Tree.id == fork.id)
)
fork = result.scalar_one()
return build_full_tree_response(fork, parent_tree=parent)
@router.get("/{tree_id}/forks", response_model=list[TreeListResponse])
async def list_forks(
tree_id: UUID,
db: Annotated[AsyncSession, Depends(get_db)],
current_user: Annotated[User, Depends(get_current_active_user)],
skip: int = Query(0, ge=0),
limit: int = Query(50, ge=1, le=100)
):
"""List all direct forks of a tree."""
# Verify parent exists and user can access it
parent_result = await db.execute(select(Tree).where(Tree.id == tree_id))
parent = parent_result.scalar_one_or_none()
if not parent:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Tree not found"
)
if not can_access_tree(current_user, parent):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You don't have access to this tree"
)
# Query direct forks, filtered by access
query = select(Tree).options(
selectinload(Tree.category_rel),
selectinload(Tree.tags)
).where(
Tree.parent_tree_id == tree_id,
Tree.is_active == True,
build_tree_access_filter(current_user)
).order_by(Tree.created_at.desc()).offset(skip).limit(limit)
result = await db.execute(query)
forks = result.scalars().unique().all()
return [build_tree_response(tree) for tree in forks]
@router.get("/{tree_id}/lineage", response_model=list[TreeListResponse])
async def get_tree_lineage(
tree_id: UUID,
db: Annotated[AsyncSession, Depends(get_db)],
current_user: Annotated[User, Depends(get_current_active_user)]
):
"""Get the fork lineage chain from current tree back to root.
Returns ordered list: [current tree, parent, grandparent, ..., root]
Limited to 10 levels to prevent infinite loops.
"""
lineage = []
current_id = tree_id
visited = set()
max_depth = 10
for _ in range(max_depth):
if current_id is None or current_id in visited:
break
visited.add(current_id)
result = await db.execute(
select(Tree)
.options(selectinload(Tree.category_rel), selectinload(Tree.tags))
.where(Tree.id == current_id)
)
tree = result.scalar_one_or_none()
if not tree:
break
lineage.append(build_tree_response(tree))
current_id = tree.parent_tree_id
return lineage
# --- Tree Sharing Endpoints ---
@router.post("/{tree_id}/share", response_model=TreeShareResponse, status_code=status.HTTP_201_CREATED)
async def create_tree_share(
tree_id: UUID,
share_data: TreeShareCreate,
request: Request,
db: Annotated[AsyncSession, Depends(get_db)],
current_user: Annotated[User, Depends(get_current_active_user)]
):
"""Generate a share token for a tree.
Requirements:
- Tree author can always create shares
- Account members can share trees with visibility 'team', 'link', or 'public'
- Super admins can share any tree
"""
# Load tree
result = await db.execute(
select(Tree).where(Tree.id == tree_id, Tree.is_active == True)
)
tree = result.scalar_one_or_none()
if not tree:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Tree not found"
)
# Check permissions
if not can_access_tree(current_user, tree):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You don't have access to this tree"
)
# Generate unique share token
share_token = secrets.token_urlsafe(48) # 48 bytes -> 64 base64 chars
# Create share
tree_share = TreeShare(
tree_id=tree.id,
share_token=share_token,
created_by=current_user.id,
allow_forking=share_data.allow_forking,
expires_at=share_data.expires_at
)
db.add(tree_share)
await log_audit(db, current_user.id, "tree.share.create", "tree_share", tree_share.id,
{"tree_id": str(tree.id), "tree_name": tree.name, "allow_forking": share_data.allow_forking})
await db.commit()
await db.refresh(tree_share)
# Build share URL
base_url = str(request.base_url).rstrip('/')
share_url = f"{base_url}/shared/{share_token}"
return TreeShareResponse(
id=tree_share.id,
tree_id=tree_share.tree_id,
share_token=tree_share.share_token,
share_url=share_url,
allow_forking=tree_share.allow_forking,
created_by=tree_share.created_by,
created_at=tree_share.created_at,
expires_at=tree_share.expires_at
)
@router.get("/{tree_id}/shares", response_model=list[TreeShareResponse])
async def list_tree_shares(
tree_id: UUID,
request: Request,
db: Annotated[AsyncSession, Depends(get_db)],
current_user: Annotated[User, Depends(get_current_active_user)]
):
"""List all active shares for a tree."""
# Verify tree exists and user can access it
result = await db.execute(
select(Tree).where(Tree.id == tree_id)
)
tree = result.scalar_one_or_none()
if not tree:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Tree not found"
)
if not can_access_tree(current_user, tree):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You don't have access to this tree"
)
# Query shares
shares_result = await db.execute(
select(TreeShare)
.where(TreeShare.tree_id == tree_id)
.order_by(TreeShare.created_at.desc())
)
shares = shares_result.scalars().all()
# Build responses with share URLs
base_url = str(request.base_url).rstrip('/')
return [
TreeShareResponse(
id=share.id,
tree_id=share.tree_id,
share_token=share.share_token,
share_url=f"{base_url}/shared/{share.share_token}",
allow_forking=share.allow_forking,
created_by=share.created_by,
created_at=share.created_at,
expires_at=share.expires_at
)
for share in shares
]
@router.patch("/{tree_id}/visibility", response_model=TreeResponse)
async def update_tree_visibility(
tree_id: UUID,
visibility_data: TreeVisibilityUpdate,
db: Annotated[AsyncSession, Depends(get_db)],
current_user: Annotated[User, Depends(require_engineer_or_admin)]
):
"""Update tree visibility level.
Visibility levels:
- private: Only tree author can access
- team: Account members can access
- link: Anyone with a valid share token can access
- public: All authenticated users can access
"""
result = await db.execute(
select(Tree).where(Tree.id == tree_id)
)
tree = result.scalar_one_or_none()
if not tree:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Tree not found"
)
if not can_edit_tree(current_user, tree):
# If the user can see this tree (same account, team visibility), give a 403 with
# a clear message — returning 404 here would be confusing since GET returns 200.
# For truly inaccessible trees (cross-account), return 404 to avoid confirming existence.
if can_access_tree(current_user, tree):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You do not have permission to edit this flow"
)
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Tree not found"
)
# Update visibility
old_visibility = tree.visibility
tree.visibility = visibility_data.visibility
tree.is_public = (visibility_data.visibility == 'public')
await log_audit(db, current_user.id, "tree.visibility.update", "tree", tree.id,
{"tree_name": tree.name, "old_visibility": old_visibility,
"new_visibility": visibility_data.visibility})
await db.commit()
# Reload with relationships
result = await db.execute(
select(Tree)
.options(
selectinload(Tree.category_rel),
selectinload(Tree.tags)
)
.where(Tree.id == tree_id)
)
tree = result.scalar_one()
return build_full_tree_response(tree)
# --- Tree Validation Endpoint ---
@router.post("/{tree_id}/can-publish", response_model=TreeValidationResponse)
async def check_tree_can_publish(
tree_id: UUID,
db: Annotated[AsyncSession, Depends(get_db)],
current_user: Annotated[User, Depends(get_current_active_user)]
):
"""Check if a tree can be published (validation endpoint).
Returns validation status and any errors that would prevent publishing.
Useful for providing real-time feedback in the UI without attempting to publish.
"""
result = await db.execute(
select(Tree).where(Tree.id == tree_id)
)
tree = result.scalar_one_or_none()
if not tree:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Tree not found"
)
if not can_access_tree(current_user, tree):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You don't have access to this tree"
)
# Validate the tree
can_publish, validation_errors = can_publish_tree(
tree.tree_structure,
tree.name,
tree.description,
tree_type=tree.tree_type,
intake_form=tree.intake_form,
)
return TreeValidationResponse(
can_publish=can_publish,
errors=[ValidationError(**error) for error in validation_errors]
)
@router.post("/{tree_id}/pin", response_model=PinnedFlowResponse)
async def pin_flow(
tree_id: UUID,
db: Annotated[AsyncSession, Depends(get_db)],
current_user: Annotated[User, Depends(get_current_active_user)]
):
"""Pin a flow to the user's sidebar."""
# Check tree exists and user can access it
tree_result = await db.execute(
select(Tree)
.options(selectinload(Tree.category_rel))
.where(Tree.id == tree_id, Tree.is_active == True)
)
tree = tree_result.scalar_one_or_none()
if not tree:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Tree not found")
if not can_access_tree(current_user, tree):
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="You don't have access to this tree")
# Check if already pinned (idempotent)
existing = await db.execute(
select(UserPinnedTree).where(
UserPinnedTree.user_id == current_user.id,
UserPinnedTree.tree_id == tree_id
)
)
pin = existing.scalar_one_or_none()
if pin:
return PinnedFlowResponse(
id=pin.id,
tree_id=tree.id,
tree_name=tree.name,
tree_type=tree.tree_type,
category_emoji=None,
category_name=tree.category_rel.name if tree.category_rel else None,
pinned_at=pin.pinned_at,
display_order=pin.display_order,
)
# Check max pins
count_result = await db.execute(
select(func.count(UserPinnedTree.id)).where(
UserPinnedTree.user_id == current_user.id
)
)
count = count_result.scalar() or 0
if count >= MAX_PINNED_FLOWS:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=f"Maximum of {MAX_PINNED_FLOWS} pinned flows reached"
)
# Create pin
pin = UserPinnedTree(
user_id=current_user.id,
tree_id=tree_id,
display_order=count, # Append at end
)
db.add(pin)
await db.commit()
await db.refresh(pin)
return PinnedFlowResponse(
id=pin.id,
tree_id=tree.id,
tree_name=tree.name,
tree_type=tree.tree_type,
category_emoji=None,
category_name=tree.category_rel.name if tree.category_rel else None,
pinned_at=pin.pinned_at,
display_order=pin.display_order,
)
@router.delete("/{tree_id}/pin")
async def unpin_flow(
tree_id: UUID,
db: Annotated[AsyncSession, Depends(get_db)],
current_user: Annotated[User, Depends(get_current_active_user)]
):
"""Unpin a flow from the user's sidebar."""
result = await db.execute(
select(UserPinnedTree).where(
UserPinnedTree.user_id == current_user.id,
UserPinnedTree.tree_id == tree_id
)
)
pin = result.scalar_one_or_none()
if pin:
await db.delete(pin)
await db.commit()
return {"success": True}