Files

1065 lines
45 KiB
Python

"""
ThinkStorm Ideas & Workflow API Router
Handles anonymous intake, public discovery, claims, work tracks, and graduation.
"""
import json
import time
from typing import Optional, List, Dict, Any
from fastapi import APIRouter, Request, HTTPException, status, Depends
from fastapi.responses import FileResponse
from pydantic import BaseModel
from ..config import config
from ..database import get_db, next_sequence, get_utc_now
from ..models import (
Idea, IdeaURL, LifecycleState, ProcessingState, User, UserRole,
WorkTrack, WorkTrackState
)
from ..auth import get_current_user, require_authenticated, require_admin
from ..queue.worker import job_queue
from ..services.gitea import GiteaAdapter
from ..services.opengist import OpenGistAdapter
from ..services.image_handler import (
validate_and_sanitize_image,
save_image_artifact,
get_image_artifact_path,
ImageValidationError
)
from ..prompts.catalog import get_all_prompts
router = APIRouter(prefix="/api/ideas", tags=["ideas"])
gitea_svc = GiteaAdapter()
opengist_svc = OpenGistAdapter()
# Rate limiting sliding window in-memory cache: ip -> list of timestamps
SUBMISSION_IP_LOG: Dict[str, List[float]] = {}
def _resolve_gitea_claimant(conn, idea, current_user: User) -> str:
"""Authorizes claimant-controlled publication and returns the Gitea username."""
claimed_by = idea["claimed_by"]
if not claimed_by:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="The idea must be claimed before it can be published to Gitea."
)
if claimed_by != current_user.username and current_user.role != UserRole.ADMIN:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Only the current claimant can publish this idea to Gitea."
)
claimant = conn.execute(
"SELECT username, gitea_id FROM users WHERE username = ?",
(claimed_by,)
).fetchone()
if not claimant or claimant["gitea_id"] is None:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="The claimant must sign in with Gitea before publishing this idea."
)
return claimant["username"]
class IdeaSubmissionRequest(BaseModel):
text: str
class IdeaClaimRequest(BaseModel):
pass
class WorkTrackCreateRequest(BaseModel):
work_type_id: str
name: str
model_override: Optional[str] = None
class WorkTrackActivateRequest(BaseModel):
model_override: Optional[str] = None
@router.post("", status_code=status.HTTP_201_CREATED)
async def submit_idea(request: Request):
"""
Public Anonymous Idea Submission.
- Zero authentication required
- Supports JSON body ({"text": "..."}) and multipart/form-data (text + optional image)
- Single image accepted, sanitized (EXIF/GPS stripped), and preserved as artifact
- Automatic TS-xxxx ID assignment
- Immutable original text preservation
- Rate-limiting & abuse prevention
"""
content_type = request.headers.get("content-type", "")
raw_text = ""
image_bytes = None
image_filename = ""
if "application/json" in content_type:
try:
body = await request.json()
if isinstance(body, dict):
raw_text = str(body.get("text") or "")
except Exception:
raise HTTPException(status_code=400, detail="Malformed JSON body.")
elif "multipart/form-data" in content_type or "application/x-www-form-urlencoded" in content_type:
try:
form = await request.form()
raw_text = str(form.get("text") or "")
img_item = form.get("image") or form.get("file")
if img_item and hasattr(img_item, "read"):
image_bytes = await img_item.read()
image_filename = getattr(img_item, "filename", "")
except Exception as e:
raise HTTPException(status_code=400, detail=f"Invalid form data: {str(e)}")
else:
# Fallback attempt JSON
try:
body = await request.json()
if isinstance(body, dict):
raw_text = str(body.get("text") or "")
except Exception:
raise HTTPException(status_code=400, detail="Invalid submission request.")
raw_text = raw_text.strip()
if not raw_text:
raise HTTPException(status_code=400, detail="Submission text cannot be empty.")
if len(raw_text) > config.max_submission_chars:
raise HTTPException(
status_code=400,
detail=f"Submission exceeds maximum allowed length of {config.max_submission_chars} characters."
)
# Rate Limiting Check
client_ip = request.client.host if request.client else "127.0.0.1"
now_ts = time.time()
recent = [t for t in SUBMISSION_IP_LOG.get(client_ip, []) if now_ts - t < config.rate_limit_window_seconds]
if len(recent) >= config.rate_limit_max_submissions:
raise HTTPException(status_code=429, detail="Rate limit exceeded. Please wait before submitting another idea.")
recent.append(now_ts)
SUBMISSION_IP_LOG[client_ip] = recent
idea_id = next_sequence("idea")
now_iso = get_utc_now()
submission_image_meta = None
if image_bytes and len(image_bytes) > 0:
try:
sanitized_bytes, submission_image_meta = validate_and_sanitize_image(
raw_bytes=image_bytes,
original_filename=image_filename,
source="web",
idea_id=idea_id
)
save_image_artifact(idea_id, sanitized_bytes, submission_image_meta)
except ImageValidationError as e:
raise HTTPException(status_code=400, detail=f"Image validation failed: {str(e)}")
except Exception as e:
raise HTTPException(status_code=400, detail=f"Could not process image upload: {str(e)}")
img_json = json.dumps(submission_image_meta) if submission_image_meta else None
with get_db() as conn:
conn.execute(
"""
INSERT INTO ideas (
id, original_text, submitted_at, title, summary,
lifecycle_state, processing_state, enrichment_level,
submission_image, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, 'SUBMITTED', 'QUEUED', 0, ?, ?, ?)
""",
(idea_id, raw_text, now_iso, "Processing New Idea...", "Analyzing submission text...", img_json, now_iso, now_iso)
)
# Enqueue intake pipeline for foreground triage
await job_queue.enqueue_foreground("intake", idea_id)
return {
"id": idea_id,
"lifecycle_state": "SUBMITTED",
"processing_state": "QUEUED",
"message": "Idea successfully accepted and queued for processing.",
"submitted_at": now_iso,
"submission_image": submission_image_meta
}
@router.get("")
async def list_ideas(
state: Optional[str] = None,
category: Optional[str] = None,
tag: Optional[str] = None,
enrichment_min: Optional[int] = None,
q: Optional[str] = None,
current_user: User = Depends(require_authenticated)
):
"""Browsing of ideas with filtering options (Requires authentication)."""
query = """
SELECT i.*,
GROUP_CONCAT(DISTINCT c.name) AS category_names,
GROUP_CONCAT(DISTINCT t.name) AS tag_names
FROM ideas i
LEFT JOIN idea_categories ic ON i.id = ic.idea_id
LEFT JOIN categories c ON ic.category_id = c.id
LEFT JOIN idea_tags it ON i.id = it.idea_id
LEFT JOIN tags t ON it.tag_id = t.id
WHERE 1=1
"""
params = []
if state and state.upper() != "ALL":
query += " AND i.lifecycle_state = ?"
params.append(state.upper())
elif state and state.upper() == "ALL":
# Show all except TRASHED
query += " AND i.lifecycle_state NOT IN ('TRASHED')"
elif not state:
# Default: show AVAILABLE, CLAIMED, ACTIVE, COMPLETED (hide QUARANTINED, REJECTED, and TRASHED)
query += " AND i.lifecycle_state NOT IN ('QUARANTINED', 'REJECTED', 'TRASHED')"
if category:
query += " AND c.name = ?"
params.append(category)
if tag:
query += " AND t.name = ?"
params.append(tag.lstrip("#").lower())
if enrichment_min is not None:
query += " AND i.enrichment_level >= ?"
params.append(enrichment_min)
if q:
query += " AND (i.title LIKE ? OR i.summary LIKE ? OR i.original_text LIKE ?)"
term = f"%{q}%"
params.extend([term, term, term])
query += " GROUP BY i.id ORDER BY i.submitted_at DESC LIMIT 50"
with get_db() as conn:
rows = conn.execute(query, params).fetchall()
results = []
for r in rows:
sub_img = None
if "submission_image" in r.keys() and r["submission_image"]:
try:
sub_img = json.loads(r["submission_image"]) if isinstance(r["submission_image"], str) else r["submission_image"]
except Exception:
sub_img = None
results.append({
"id": r["id"],
"title": r["title"] or "Untitled Idea",
"summary": r["summary"],
"lifecycle_state": r["lifecycle_state"],
"processing_state": r["processing_state"],
"enrichment_level": r["enrichment_level"],
"claimed_by": r["claimed_by"],
"submitted_at": r["submitted_at"],
"categories": [c.strip() for c in r["category_names"].split(",")] if r["category_names"] else [],
"tags": [t.strip() for t in r["tag_names"].split(",")] if r["tag_names"] else [],
"submission_image": sub_img
})
return results
@router.post("/trash/empty")
@router.delete("/trash")
async def empty_trash(current_user: User = Depends(get_current_user)):
"""Permanently deletes all ideas currently in TRASHED state along with all their cascading records."""
with get_db() as conn:
trashed_rows = conn.execute("SELECT id FROM ideas WHERE lifecycle_state = 'TRASHED'").fetchall()
trashed_ids = [r["id"] for r in trashed_rows]
if not trashed_ids:
return {"deleted_count": 0, "message": "Trash is already empty."}
for i_id in trashed_ids:
conn.execute("DELETE FROM idea_categories WHERE idea_id = ?", (i_id,))
conn.execute("DELETE FROM idea_tags WHERE idea_id = ?", (i_id,))
conn.execute("DELETE FROM idea_urls WHERE idea_id = ?", (i_id,))
conn.execute("DELETE FROM idea_relationships WHERE source_idea_id = ? OR target_idea_id = ?", (i_id, i_id))
conn.execute("DELETE FROM processor_runs WHERE idea_id = ?", (i_id,))
conn.execute("DELETE FROM work_track_outputs WHERE work_track_id IN (SELECT id FROM work_tracks WHERE idea_id = ?)", (i_id,))
conn.execute("DELETE FROM external_resources WHERE idea_id = ?", (i_id,))
conn.execute("DELETE FROM work_tracks WHERE idea_id = ?", (i_id,))
conn.execute("DELETE FROM ideas WHERE id = ?", (i_id,))
now = get_utc_now()
conn.execute(
"""
INSERT INTO audit_events (user_id, action, entity_type, entity_id, details, created_at)
VALUES (?, 'EMPTY_TRASH', 'TRASH', 'ALL', ?, ?)
""",
(current_user.username, json.dumps({"deleted_count": len(trashed_ids), "deleted_ids": trashed_ids}), now)
)
return {
"deleted_count": len(trashed_ids),
"deleted_ids": trashed_ids,
"message": f"Successfully emptied trash. {len(trashed_ids)} idea(s) permanently deleted."
}
@router.get("/{idea_id}")
async def get_idea_detail(idea_id: str, current_user: User = Depends(require_authenticated)):
"""Fetches full idea dossier with URLs, research docs, work tracks, and provenance (Requires authentication)."""
is_admin = current_user.role == UserRole.ADMIN
with get_db() as conn:
idea_row = conn.execute("SELECT * FROM ideas WHERE id = ?", (idea_id,)).fetchone()
if not idea_row:
raise HTTPException(status_code=404, detail="Idea not found.")
# If quarantined, non-admins cannot view
if idea_row["lifecycle_state"] == "QUARANTINED" and not is_admin:
raise HTTPException(status_code=403, detail="This idea is currently under administrative safety review.")
# URLs
url_rows = conn.execute("SELECT * FROM idea_urls WHERE idea_id = ?", (idea_id,)).fetchall()
urls = [
{
"id": u["id"],
"url": u["url"],
"safety_state": u["safety_state"],
"automation_policy": u["automation_policy"],
"virustotal": json.loads(u["virustotal_data"] or "{}")
}
for u in url_rows
]
# Categories & Tags
cats = [r["name"] for r in conn.execute("SELECT c.name FROM categories c JOIN idea_categories ic ON c.id = ic.category_id WHERE ic.idea_id = ?", (idea_id,)).fetchall()]
tags = [r["name"] for r in conn.execute("SELECT t.name FROM tags t JOIN idea_tags it ON t.id = it.tag_id WHERE it.idea_id = ?", (idea_id,)).fetchall()]
track_rows = conn.execute("SELECT * FROM work_tracks WHERE idea_id = ?", (idea_id,)).fetchall()
work_tracks = []
for tr in track_rows:
output_rows = conn.execute(
"SELECT * FROM work_track_outputs WHERE work_track_id = ? ORDER BY name ASC, version DESC",
(tr["id"],)
).fetchall()
ext_rows = conn.execute("SELECT * FROM external_resources WHERE work_track_id = ?", (tr["id"],)).fetchall()
work_tracks.append({
"id": tr["id"],
"work_type_id": tr["work_type_id"],
"name": tr["name"],
"state": tr["state"],
"workflow_id": tr["workflow_id"],
"model_override": tr["model_override"] if "model_override" in tr.keys() else None,
"created_at": tr["created_at"],
"started_at": tr["started_at"],
"completed_at": tr["completed_at"],
"outputs": [
{
"id": o["id"],
"name": o["name"],
"artifact_path": o["artifact_path"],
"content": o["content"],
"version": o["version"] if "version" in o.keys() else 1,
"is_current": bool(o["is_current"]) if "is_current" in o.keys() else True,
"model_used": o["model_used"] if "model_used" in o.keys() else None,
"created_at": o["created_at"]
}
for o in output_rows
],
"external_resources": [{"type": e["resource_type"], "url": e["url"]} for e in ext_rows]
})
# Provenance runs (with role-based prompt masking)
run_rows = conn.execute("SELECT * FROM processor_runs WHERE idea_id = ? ORDER BY started_at ASC", (idea_id,)).fetchall()
provenance = []
total_tokens_sum = 0
for rn in run_rows:
in_tok = rn["input_tokens"]
out_tok = rn["output_tokens"]
tot_tok = rn["total_tokens"]
total_tokens_sum += tot_tok
provenance.append({
"id": rn["id"],
"processor_name": rn["processor_name"],
"stage": rn["stage"],
"prompt_id": rn["prompt_id"],
"prompt_version": rn["prompt_version"],
"prompt_hash": rn["prompt_hash"],
"model_policy": rn["model_policy"],
"resolved_provider": rn["resolved_provider"],
"resolved_model": rn["resolved_model"],
"input_tokens": in_tok,
"output_tokens": out_tok,
"total_tokens": tot_tok,
"started_at": rn["started_at"],
"completed_at": rn["completed_at"],
"duration_ms": rn["duration_ms"],
"output_artifact": rn["output_artifact"],
"output_data": json.loads(rn["output_data"] or "{}"),
"status": rn["status"]
})
# Relationships
rels = [
{"target_idea_id": r["target_idea_id"], "relationship_type": r["relationship_type"], "notes": r["notes"]}
for r in conn.execute("SELECT * FROM idea_relationships WHERE source_idea_id = ?", (idea_id,)).fetchall()
]
# Image metadata
sub_img = None
if "submission_image" in idea_row.keys() and idea_row["submission_image"]:
try:
sub_img = json.loads(idea_row["submission_image"]) if isinstance(idea_row["submission_image"], str) else idea_row["submission_image"]
except Exception:
sub_img = None
# Build multi-version research stages
stage_definitions = {
"IMAGE_CONTEXT": {"name": "Visual & Image Context", "doc_name": "image-context.md", "icon": "🖼️"},
"PRIOR_ART": {"name": "Prior Art & Alternatives", "doc_name": "prior-art.md", "icon": "🔍"},
"RESEARCH": {"name": "Deep Research Synthesis", "doc_name": "analysis.md", "icon": "📚"},
"FEASIBILITY": {"name": "Feasibility & Critique", "doc_name": "feasibility.md", "icon": "⚙️"}
}
research_stages = {}
for stage_key, defn in stage_definitions.items():
stage_runs = [
rn for rn in provenance
if rn["stage"] == stage_key and rn["output_data"] and rn["output_data"].get("content")
]
stage_runs_sorted = sorted(stage_runs, key=lambda x: x["started_at"] or "")
v_list = []
for idx, r_item in enumerate(stage_runs_sorted, start=1):
v_obj = {
"version": idx,
"is_latest": False,
"run_id": r_item["id"],
"processor_name": r_item["processor_name"],
"model": r_item["resolved_model"] or "Default Policy",
"provider": r_item["resolved_provider"] or "",
"tokens": {
"input": r_item["input_tokens"],
"output": r_item["output_tokens"],
"total": r_item["total_tokens"]
},
"started_at": r_item["started_at"],
"completed_at": r_item["completed_at"],
"duration_ms": r_item["duration_ms"],
"prompt_id": r_item["prompt_id"],
"prompt_version": r_item["prompt_version"],
"artifact_path": r_item["output_artifact"] or f"research/{defn['doc_name']}",
"content": r_item["output_data"].get("content", "")
}
v_list.append(v_obj)
if v_list:
v_list[-1]["is_latest"] = True
latest_version = v_list[-1]
else:
latest_version = None
research_stages[stage_key] = {
"stage": stage_key,
"name": defn["name"],
"doc_name": defn["doc_name"],
"icon": defn["icon"],
"versions": v_list,
"latest": latest_version,
"total_versions": len(v_list)
}
claimant_gitea_linked = False
if idea_row["claimed_by"]:
claimant_row = conn.execute(
"SELECT gitea_id FROM users WHERE username = ?",
(idea_row["claimed_by"],)
).fetchone()
claimant_gitea_linked = bool(
claimant_row and claimant_row["gitea_id"] is not None
)
actor_is_claimant = idea_row["claimed_by"] == current_user.username
gitea_publish_allowed = bool(
idea_row["claimed_by"]
and claimant_gitea_linked
and (actor_is_claimant or is_admin)
)
if not idea_row["claimed_by"]:
gitea_publish_reason = "Claim this idea before publishing it to Gitea."
elif not actor_is_claimant and not is_admin:
gitea_publish_reason = "Only the current claimant can publish this idea."
elif not claimant_gitea_linked:
gitea_publish_reason = "The claimant must sign in with Gitea before publishing."
else:
gitea_publish_reason = ""
return {
"id": idea_row["id"],
"title": idea_row["title"],
"summary": idea_row["summary"],
"original_text": idea_row["original_text"],
"submitted_at": idea_row["submitted_at"],
"lifecycle_state": idea_row["lifecycle_state"],
"processing_state": idea_row["processing_state"],
"enrichment_level": idea_row["enrichment_level"],
"claimed_by": idea_row["claimed_by"],
"claimed_at": idea_row["claimed_at"],
"released_at": idea_row["released_at"],
"previous_lifecycle_state": idea_row["previous_lifecycle_state"],
"trashed_at": idea_row["trashed_at"],
"profile_id": idea_row["profile_id"],
"opengist_id": idea_row["opengist_id"],
"opengist_url": idea_row["opengist_url"],
"gitea_repo_name": idea_row["gitea_repo_name"] if "gitea_repo_name" in idea_row.keys() else "",
"gitea_repo_url": idea_row["gitea_repo_url"] if "gitea_repo_url" in idea_row.keys() else "",
"gitea_publish_allowed": gitea_publish_allowed,
"gitea_publish_reason": gitea_publish_reason,
"submission_image": sub_img,
"categories": cats,
"tags": tags,
"urls": urls,
"work_tracks": work_tracks,
"provenance": provenance,
"research_stages": research_stages,
"relationships": rels,
"usage_summary": {
"total_tokens": total_tokens_sum,
"runs_count": len(provenance)
}
}
@router.post("/{idea_id}/claim")
async def claim_idea(idea_id: str, current_user: User = Depends(require_authenticated)):
"""Authenticated user claims an AVAILABLE idea."""
now = get_utc_now()
with get_db() as conn:
idea = conn.execute("SELECT * FROM ideas WHERE id = ?", (idea_id,)).fetchone()
if not idea:
raise HTTPException(status_code=404, detail="Idea not found.")
if idea["lifecycle_state"] != "AVAILABLE":
raise HTTPException(status_code=400, detail=f"Idea in '{idea['lifecycle_state']}' state cannot be claimed.")
conn.execute(
"""
UPDATE ideas
SET lifecycle_state = 'CLAIMED', claimed_by = ?, claimed_at = ?, updated_at = ?
WHERE id = ?
""",
(current_user.username, now, now, idea_id)
)
return {"message": f"Idea {idea_id} successfully claimed by {current_user.username}."}
@router.post("/{idea_id}/release")
async def release_claim(idea_id: str, current_user: User = Depends(require_authenticated)):
"""Claimant or admin releases a claimed idea back to AVAILABLE."""
now = get_utc_now()
with get_db() as conn:
idea = conn.execute("SELECT * FROM ideas WHERE id = ?", (idea_id,)).fetchone()
if not idea:
raise HTTPException(status_code=404, detail="Idea not found.")
if idea["claimed_by"] != current_user.username and current_user.role != UserRole.ADMIN:
raise HTTPException(status_code=403, detail="Only the claimant or an administrator can release this claim.")
conn.execute(
"""
UPDATE ideas
SET lifecycle_state = 'AVAILABLE', claimed_by = NULL, released_at = ?, updated_at = ?
WHERE id = ?
""",
(now, now, idea_id)
)
return {"message": f"Idea {idea_id} claim released and returned to AVAILABLE."}
@router.post("/{idea_id}/activate")
async def activate_idea(idea_id: str, current_user: User = Depends(require_authenticated)):
"""Explicitly transitions a CLAIMED idea to ACTIVE."""
now = get_utc_now()
with get_db() as conn:
idea = conn.execute("SELECT * FROM ideas WHERE id = ?", (idea_id,)).fetchone()
if not idea:
raise HTTPException(status_code=404, detail="Idea not found.")
if idea["claimed_by"] != current_user.username and current_user.role != UserRole.ADMIN:
raise HTTPException(status_code=403, detail="Only the claimant can activate work.")
conn.execute(
"UPDATE ideas SET lifecycle_state = 'ACTIVE', updated_at = ? WHERE id = ?",
(now, idea_id)
)
return {"message": f"Idea {idea_id} is now ACTIVE."}
@router.post("/{idea_id}/work-tracks")
async def create_work_track(
idea_id: str,
payload: WorkTrackCreateRequest,
current_user: User = Depends(require_authenticated)
):
"""Creates a new independent work track for a claimed idea."""
now = get_utc_now()
with get_db() as conn:
idea = conn.execute("SELECT * FROM ideas WHERE id = ?", (idea_id,)).fetchone()
if not idea:
raise HTTPException(status_code=404, detail="Idea not found.")
if idea["claimed_by"] != current_user.username and current_user.role != UserRole.ADMIN:
raise HTTPException(status_code=403, detail="Only the claimant can create work tracks.")
# Check work type
wt = conn.execute("SELECT * FROM work_types WHERE id = ? AND enabled = 1", (payload.work_type_id,)).fetchone()
if not wt:
raise HTTPException(status_code=400, detail=f"Work type '{payload.work_type_id}' is invalid or disabled.")
track_id = next_sequence("work_track")
conn.execute(
"""
INSERT INTO work_tracks (id, idea_id, work_type_id, name, state, workflow_id, model_override, created_at)
VALUES (?, ?, ?, ?, 'PLANNED', ?, ?, ?)
""",
(track_id, idea_id, payload.work_type_id, payload.name, wt["default_workflow_id"], payload.model_override, now)
)
return {
"id": track_id,
"work_type_id": payload.work_type_id,
"name": payload.name,
"state": "PLANNED",
"model_override": payload.model_override,
"created_at": now
}
@router.post("/work-tracks/{track_id}/activate")
async def activate_work_track(
track_id: str,
payload: Optional[WorkTrackActivateRequest] = None,
current_user: User = Depends(get_current_user)
):
"""Deliberately begins work track workflow execution with optional model override."""
with get_db() as conn:
track = conn.execute("SELECT * FROM work_tracks WHERE id = ?", (track_id,)).fetchone()
if not track:
raise HTTPException(status_code=404, detail="Work track not found.")
idea = conn.execute("SELECT * FROM ideas WHERE id = ?", (track["idea_id"],)).fetchone()
if idea["claimed_by"] and idea["claimed_by"] != current_user.username and current_user.role != UserRole.ADMIN:
raise HTTPException(status_code=403, detail="Only the claimant can activate this track.")
if payload and payload.model_override:
conn.execute("UPDATE work_tracks SET model_override = ? WHERE id = ?", (payload.model_override, track_id))
await job_queue.enqueue_foreground("work_track", track_id)
return {"message": f"Work track {track_id} queued for activation and generation."}
@router.post("/work-tracks/{track_id}/graduate")
async def graduate_work_track_to_gitea(track_id: str, current_user: User = Depends(require_authenticated)):
"""Graduates an incubated Coding Project work track to Gitea."""
with get_db() as conn:
track = conn.execute("SELECT * FROM work_tracks WHERE id = ?", (track_id,)).fetchone()
if not track:
raise HTTPException(status_code=404, detail="Work track not found.")
if track["work_type_id"] != "CODING_PROJECT":
raise HTTPException(status_code=400, detail="Only Coding Project work tracks can graduate to Gitea.")
idea = conn.execute("SELECT * FROM ideas WHERE id = ?", (track["idea_id"],)).fetchone()
claimant_username = _resolve_gitea_claimant(conn, idea, current_user)
try:
res = await gitea_svc.graduate_project(
idea["id"],
idea["title"],
idea["summary"],
collaborator_username=claimant_username
)
except RuntimeError as exc:
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY,
detail=str(exc)
) from exc
now = get_utc_now()
with get_db() as conn:
conn.execute(
"""
INSERT INTO external_resources (idea_id, work_track_id, resource_type, url, metadata, created_at)
VALUES (?, ?, 'GITEA_REPOSITORY', ?, ?, ?)
""",
(idea["id"], track_id, res["repo_url"], json.dumps(res), now)
)
return {
"message": "Coding project successfully graduated to Gitea.",
"repo_name": res["repo_name"],
"repo_url": res["repo_url"]
}
@router.delete("/work-tracks/outputs/{output_id}")
async def delete_work_track_output(output_id: int, current_user: User = Depends(get_current_user)):
"""Deletes a specific generated artifact version from a work track."""
with get_db() as conn:
out = conn.execute(
"""
SELECT wto.*, wt.idea_id, i.claimed_by
FROM work_track_outputs wto
JOIN work_tracks wt ON wto.work_track_id = wt.id
JOIN ideas i ON wt.idea_id = i.id
WHERE wto.id = ?
""",
(output_id,)
).fetchone()
if not out:
raise HTTPException(status_code=404, detail="Artifact not found.")
if out["claimed_by"] and out["claimed_by"] != current_user.username and current_user.role != UserRole.ADMIN:
raise HTTPException(status_code=403, detail="Only the claimant or admin can delete artifacts.")
track_id = out["work_track_id"]
name = out["name"]
is_curr = out["is_current"]
ver = out["version"]
conn.execute("DELETE FROM work_track_outputs WHERE id = ?", (output_id,))
# If the deleted artifact was marked is_current, promote the highest remaining version
if is_curr:
next_top = conn.execute(
"SELECT id FROM work_track_outputs WHERE work_track_id = ? AND name = ? ORDER BY version DESC LIMIT 1",
(track_id, name)
).fetchone()
if next_top:
conn.execute("UPDATE work_track_outputs SET is_current = 1 WHERE id = ?", (next_top["id"],))
return {"message": f"Artifact {name} (v{ver}) deleted successfully."}
# -------------------------------------------------------------
# Idea Trash, Restore & Permanent Deletion
# -------------------------------------------------------------
@router.post("/{idea_id}/trash")
async def trash_idea(idea_id: str, current_user: User = Depends(get_current_user)):
"""Moves an idea to TRASHED state while preserving its previous lifecycle state."""
now = get_utc_now()
with get_db() as conn:
idea = conn.execute("SELECT * FROM ideas WHERE id = ?", (idea_id,)).fetchone()
if not idea:
raise HTTPException(status_code=404, detail="Idea not found.")
if idea["lifecycle_state"] == "TRASHED":
return {"message": f"Idea {idea_id} is already in the trash.", "id": idea_id, "lifecycle_state": "TRASHED"}
prev_state = idea["lifecycle_state"]
conn.execute(
"""
UPDATE ideas
SET lifecycle_state = 'TRASHED',
previous_lifecycle_state = ?,
trashed_at = ?,
updated_at = ?
WHERE id = ?
""",
(prev_state, now, now, idea_id)
)
conn.execute(
"""
INSERT INTO audit_events (user_id, action, entity_type, entity_id, details, created_at)
VALUES (?, 'TRASH_IDEA', 'IDEA', ?, ?, ?)
""",
(current_user.username, idea_id, json.dumps({"previous_state": prev_state}), now)
)
return {
"message": f"Idea {idea_id} moved to trash.",
"id": idea_id,
"lifecycle_state": "TRASHED",
"previous_lifecycle_state": prev_state
}
@router.post("/{idea_id}/restore")
async def restore_idea(idea_id: str, current_user: User = Depends(get_current_user)):
"""Restores a TRASHED idea back to its previous lifecycle state."""
now = get_utc_now()
with get_db() as conn:
idea = conn.execute("SELECT * FROM ideas WHERE id = ?", (idea_id,)).fetchone()
if not idea:
raise HTTPException(status_code=404, detail="Idea not found.")
if idea["lifecycle_state"] != "TRASHED":
raise HTTPException(status_code=400, detail=f"Idea {idea_id} is not in the trash (current state: '{idea['lifecycle_state']}').")
prev_state = idea["previous_lifecycle_state"]
if not prev_state or prev_state in ("TRASHED", "SUBMITTED"):
restore_target = "AVAILABLE"
else:
restore_target = prev_state
conn.execute(
"""
UPDATE ideas
SET lifecycle_state = ?,
trashed_at = NULL,
updated_at = ?
WHERE id = ?
""",
(restore_target, now, idea_id)
)
conn.execute(
"""
INSERT INTO audit_events (user_id, action, entity_type, entity_id, details, created_at)
VALUES (?, 'RESTORE_IDEA', 'IDEA', ?, ?, ?)
""",
(current_user.username, idea_id, json.dumps({"restored_to": restore_target}), now)
)
return {
"message": f"Idea {idea_id} successfully restored to '{restore_target}'.",
"id": idea_id,
"lifecycle_state": restore_target
}
@router.delete("/{idea_id}/permanent")
@router.delete("/{idea_id}")
async def delete_idea_permanently(idea_id: str, current_user: User = Depends(get_current_user)):
"""Permanently deletes a single idea and all related records."""
with get_db() as conn:
idea = conn.execute("SELECT * FROM ideas WHERE id = ?", (idea_id,)).fetchone()
if not idea:
raise HTTPException(status_code=404, detail="Idea not found.")
# Cascade delete all related records
conn.execute("DELETE FROM idea_categories WHERE idea_id = ?", (idea_id,))
conn.execute("DELETE FROM idea_tags WHERE idea_id = ?", (idea_id,))
conn.execute("DELETE FROM idea_urls WHERE idea_id = ?", (idea_id,))
conn.execute("DELETE FROM idea_relationships WHERE source_idea_id = ? OR target_idea_id = ?", (idea_id, idea_id))
conn.execute("DELETE FROM processor_runs WHERE idea_id = ?", (idea_id,))
conn.execute("DELETE FROM work_track_outputs WHERE work_track_id IN (SELECT id FROM work_tracks WHERE idea_id = ?)", (idea_id,))
conn.execute("DELETE FROM external_resources WHERE idea_id = ?", (idea_id,))
conn.execute("DELETE FROM work_tracks WHERE idea_id = ?", (idea_id,))
conn.execute("DELETE FROM ideas WHERE id = ?", (idea_id,))
now = get_utc_now()
conn.execute(
"""
INSERT INTO audit_events (user_id, action, entity_type, entity_id, details, created_at)
VALUES (?, 'DELETE_PERMANENT', 'IDEA', ?, '{}', ?)
""",
(current_user.username, idea_id, now)
)
return {
"message": f"Idea {idea_id} permanently deleted.",
"id": idea_id
}
@router.post("/{idea_id}/sync-opengist")
async def sync_idea_to_opengist(idea_id: str, current_user: User = Depends(get_current_user)):
"""Manually triggers full OpenGist sync for an idea and all its artifacts."""
with get_db() as conn:
idea = conn.execute("SELECT * FROM ideas WHERE id = ?", (idea_id,)).fetchone()
if not idea:
raise HTTPException(status_code=404, detail="Idea not found.")
cats = [r["name"] for r in conn.execute("SELECT c.name FROM categories c JOIN idea_categories ic ON c.id = ic.category_id WHERE ic.idea_id = ?", (idea_id,)).fetchall()]
tags = [r["name"] for r in conn.execute("SELECT t.name FROM tags t JOIN idea_tags it ON t.id = it.tag_id WHERE it.idea_id = ?", (idea_id,)).fetchall()]
runs = [dict(r) for r in conn.execute("SELECT * FROM processor_runs WHERE idea_id = ? ORDER BY started_at ASC", (idea_id,)).fetchall()]
# Build research docs (chronologically latest run content takes precedence)
research_docs = {}
for r in runs:
stage = r.get("stage")
out_raw = r.get("output_data") or "{}"
try:
out_data = json.loads(out_raw) if isinstance(out_raw, str) else out_raw
except Exception:
out_data = {}
content = out_data.get("content")
if content:
if stage == "IMAGE_CONTEXT":
research_docs["image-context.md"] = content
elif stage == "PRIOR_ART":
research_docs["prior-art.md"] = content
elif stage == "RESEARCH":
research_docs["analysis.md"] = content
elif stage == "FEASIBILITY":
research_docs["feasibility.md"] = content
res = await opengist_svc.persist_idea_artifact(
idea_id=idea["id"],
title=idea["title"] or "Untitled Idea",
summary=idea["summary"] or "",
original_text=idea["original_text"] or "",
categories=cats,
tags=tags,
lifecycle_state=idea["lifecycle_state"],
research_docs=research_docs,
outputs={},
provenance_runs=runs,
existing_gist_id=idea["opengist_id"]
)
with get_db() as conn:
conn.execute(
"""
UPDATE ideas
SET opengist_id = ?, opengist_url = ?, updated_at = ?
WHERE id = ?
""",
(res["opengist_id"], res["opengist_url"], get_utc_now(), idea_id)
)
return {
"message": "Dossier successfully published and synced to OpenGist.",
"opengist_id": res["opengist_id"],
"opengist_url": res["opengist_url"]
}
@router.post("/{idea_id}/sync-gitea")
async def sync_idea_to_gitea(idea_id: str, current_user: User = Depends(require_authenticated)):
"""Publishes or re-syncs a claimed idea and grants its claimant write access."""
with get_db() as conn:
idea = conn.execute("SELECT * FROM ideas WHERE id = ?", (idea_id,)).fetchone()
if not idea:
raise HTTPException(status_code=404, detail="Idea not found.")
claimant_username = _resolve_gitea_claimant(conn, idea, current_user)
cats = [r["name"] for r in conn.execute("SELECT c.name FROM categories c JOIN idea_categories ic ON c.id = ic.category_id WHERE ic.idea_id = ?", (idea_id,)).fetchall()]
tags = [r["name"] for r in conn.execute("SELECT t.name FROM tags t JOIN idea_tags it ON t.id = it.tag_id WHERE it.idea_id = ?", (idea_id,)).fetchall()]
runs = [dict(r) for r in conn.execute("SELECT * FROM processor_runs WHERE idea_id = ? ORDER BY started_at ASC", (idea_id,)).fetchall()]
# Build research docs (chronologically latest run content takes precedence)
research_docs = {}
for r in runs:
stage = r.get("stage")
out_raw = r.get("output_data") or "{}"
try:
out_data = json.loads(out_raw) if isinstance(out_raw, str) else out_raw
except Exception:
out_data = {}
content = out_data.get("content")
if content:
if stage == "IMAGE_CONTEXT":
research_docs["image-context.md"] = content
elif stage == "PRIOR_ART":
research_docs["prior-art.md"] = content
elif stage == "RESEARCH":
research_docs["analysis.md"] = content
elif stage == "FEASIBILITY":
research_docs["feasibility.md"] = content
# Collect work track outputs
outputs = {}
track_rows = conn.execute("SELECT id, name FROM work_tracks WHERE idea_id = ?", (idea_id,)).fetchall()
for tr in track_rows:
tr_outputs = conn.execute(
"SELECT name, content, version FROM work_track_outputs WHERE work_track_id = ? AND is_current = 1",
(tr["id"],)
).fetchall()
for out in tr_outputs:
track_slug = tr["name"].lower().replace(" ", "-")
outputs[f"{track_slug}/{out['name']}"] = out["content"]
try:
res = await gitea_svc.persist_idea_dossier_repo(
idea_id=idea["id"],
title=idea["title"] or "Untitled Idea",
summary=idea["summary"] or "",
original_text=idea["original_text"] or "",
categories=cats,
tags=tags,
lifecycle_state=idea["lifecycle_state"],
research_docs=research_docs,
outputs=outputs,
provenance_runs=runs,
existing_repo_url=(
idea["gitea_repo_url"]
if "gitea_repo_url" in idea.keys() else None
),
publish_remote=True,
collaborator_username=claimant_username
)
except RuntimeError as exc:
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY,
detail=str(exc)
) from exc
now = get_utc_now()
with get_db() as conn:
conn.execute(
"""
UPDATE ideas
SET gitea_repo_name = ?, gitea_repo_url = ?, updated_at = ?
WHERE id = ?
""",
(res["repo_name"], res["repo_url"], now, idea_id)
)
existing_resource = conn.execute(
"""
SELECT id FROM external_resources
WHERE idea_id = ? AND resource_type = 'GITEA_DOSSIER'
ORDER BY id ASC LIMIT 1
""",
(idea_id,)
).fetchone()
if existing_resource:
conn.execute(
"""
UPDATE external_resources
SET url = ?, metadata = ?
WHERE id = ?
""",
(res["repo_url"], json.dumps(res), existing_resource["id"])
)
else:
conn.execute(
"""
INSERT INTO external_resources
(idea_id, resource_type, url, metadata, created_at)
VALUES (?, 'GITEA_DOSSIER', ?, ?, ?)
""",
(idea_id, res["repo_url"], json.dumps(res), now)
)
return {
"message": (
f"Dossier published to '{res['repo_name']}' and write access granted "
f"to '{claimant_username}'."
),
"gitea_repo_name": res["repo_name"],
"gitea_repo_url": res["repo_url"],
"clone_url": res["clone_url"],
"ssh_url": res["ssh_url"]
}
@router.post("/{idea_id}/reprocess")
async def reprocess_idea(
idea_id: str,
bypass_duplicate_check: bool = True,
current_user: User = Depends(get_current_user)
):
"""Triggers research reprocessing without automatically publishing to Gitea."""
with get_db() as conn:
idea = conn.execute("SELECT * FROM ideas WHERE id = ?", (idea_id,)).fetchone()
if not idea:
raise HTTPException(status_code=404, detail="Idea not found.")
now = get_utc_now()
conn.execute(
"UPDATE ideas SET processing_state = 'QUEUED', updated_at = ? WHERE id = ?",
(now, idea_id)
)
await job_queue.enqueue_foreground("intake", idea_id, {"bypass_duplicate_check": bypass_duplicate_check})
return {
"message": f"Idea {idea_id} enqueued for processing.",
"idea_id": idea_id,
"processing_state": "QUEUED"
}
@router.get("/{idea_id}/image")
async def get_idea_image(idea_id: str):
"""Serves the canonical reference image attached to an idea."""
with get_db() as conn:
row = conn.execute("SELECT submission_image FROM ideas WHERE id = ?", (idea_id,)).fetchone()
if not row:
raise HTTPException(status_code=404, detail="Idea not found.")
raw_img = row["submission_image"] if "submission_image" in row.keys() else None
if not raw_img:
raise HTTPException(status_code=404, detail="No reference image associated with this idea.")
meta = json.loads(raw_img) if isinstance(raw_img, str) else raw_img
if not meta or not meta.get("present"):
raise HTTPException(status_code=404, detail="No reference image associated with this idea.")
img_path = get_image_artifact_path(idea_id, meta)
if not img_path or not img_path.exists():
raise HTTPException(status_code=404, detail="Image artifact file not found on disk.")
mime = meta.get("mime_type", "image/jpeg")
return FileResponse(
path=str(img_path),
media_type=mime,
filename=meta.get("original_filename", f"{idea_id}_reference_image")
)