Unify project state management: DB as source of truth for config/status

- Extended Project DB schema with config fields (model, code_dir, language,
  source), runtime status (phase, iteration, dev_iteration, score_history),
  cost tracking (total_cost_usd, total_input/output_tokens), checkpoint
  data, and process tracking (pid)
- Made orchestrator DB-aware: _sync_db() updates project record after
  each step, checkpoint, score change, and cost report write
- Unified CLI with webapp: ark new registers projects in webapp DB,
  ark run passes --db-path/--project-id to orchestrator
- Updated webapp to read from DB first with YAML fallback for legacy
  projects. SSE stream now reads from DB instead of parsing YAML
  every 2 seconds
- Added data migration (migrate_project_data) that populates new DB
  columns from existing YAML state files on webapp startup
- YAML files retained for agent working state (findings, experiment_plan,
  research_state, action_plan, literature, memory)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: JihaoXin

ark/cli.py

@@ -1591,6 +1591,47 @@ def _finalize_project(name: str, project_dir: Path, config: dict,
     """Create project directory, write config, copy templates, set up workspace."""
     code_dir = config["code_dir"]
 
+    # ── Register project in webapp DB ──────────────────────────
+    project_id = None
+    try:
+        from ark.webapp.db import (resolve_db_path, get_session,
+                                   get_or_create_user_by_email, create_project as db_create_project)
+        import getpass
+        db_path = resolve_db_path()
+        if db_path and Path(db_path).parent.exists():
+            with get_session(db_path) as session:
+                cli_email = f"{getpass.getuser()}@cli.local"
+                user, _ = get_or_create_user_by_email(session, cli_email)
+                import uuid
+                project_id = str(uuid.uuid4())
+                db_create_project(
+                    session,
+                    id=project_id,
+                    user_id=user.id,
+                    name=title or name,
+                    title=title or "",
+                    idea=config.get("research_idea", "") or config.get("idea", ""),
+                    venue=venue_name,
+                    venue_format=venue_format,
+                    venue_pages=venue_pages,
+                    max_iterations=config.get("max_iterations", 2),
+                    max_dev_iterations=config.get("max_dev_iterations", 3),
+                    mode=config.get("mode", "paper"),
+                    model=config.get("model", "claude"),
+                    model_variant=config.get("model_variant", ""),
+                    code_dir=str(code_dir),
+                    language=config.get("language", "en"),
+                    source="cli",
+                    status="queued",
+                    telegram_token=config.get("telegram_bot_token", ""),
+                    telegram_chat_id=config.get("telegram_chat_id", ""),
+                )
+            # Store project_id in config so cmd_run can pass it to orchestrator
+            config["_project_id"] = project_id
+            config["_db_path"] = db_path
+    except Exception as e:
+        print(f"  {_c('Note:', Colors.DIM)} DB registration skipped: {e}")
+
     # ── Create project directory ──────────────────────────────
     project_dir.mkdir(parents=True, exist_ok=True)
     agents_dir = project_dir / "agents"
@@ -1830,6 +1871,25 @@ def cmd_run(args):
     print(f"  Max days:       {max_days}")
     print(f"  Log file:       {log_file}")
 
+    # ── Resolve DB path and project ID for orchestrator ──
+    db_path = config.get("_db_path", "")
+    project_id = config.get("_project_id", "")
+    if not db_path:
+        try:
+            from ark.webapp.db import resolve_db_path
+            db_path = resolve_db_path()
+        except Exception:
+            pass
+    if not project_id and db_path and Path(db_path).exists():
+        try:
+            from ark.webapp.db import get_session, get_project_by_name
+            with get_session(db_path) as session:
+                p = get_project_by_name(session, name)
+                if p:
+                    project_id = p.id
+        except Exception:
+            pass
+
     # Launch orchestrator in background
     cmd = [
         sys.executable, "-m", "ark.orchestrator",
@@ -1839,6 +1899,10 @@ def cmd_run(args):
         "--iterations", str(max_iterations),
         "--max-days", str(max_days),
     ]
+    if db_path:
+        cmd.extend(["--db-path", db_path])
+    if project_id:
+        cmd.extend(["--project-id", project_id])
 
     # Strip CLAUDECODE so orchestrator can call claude CLI freely
     env = {k: v for k, v in os.environ.items() if k != "CLAUDECODE"}

ark/orchestrator.py

@@ -44,7 +44,7 @@ class Orchestrator(AgentMixin, CompilerMixin, ExecutionMixin, PipelineMixin, Dev
 
     def __init__(self, project: str, max_days: float = 3, max_iterations: int = 100,
                  mode: str = "research", model: str = "claude", code_dir: str = None,
-                 project_dir: str = None):
+                 project_dir: str = None, db_path: str = None, project_id: str = None):
         global PROJECT_DIR
 
         self.max_end_time = datetime.now() + timedelta(days=max_days)
@@ -54,6 +54,11 @@ def __init__(self, project: str, max_days: float = 3, max_iterations: int = 100,
         self.model = model
         self.project_name = project
 
+        # ── DB awareness ──
+        self._db_path = db_path
+        self._project_id = project_id
+        self._db_sync_errors = 0  # count consecutive failures
+
         # Human-readable display name (resolved after config loads)
         self._display_name = None
 
@@ -98,7 +103,7 @@ def __init__(self, project: str, max_days: float = 3, max_iterations: int = 100,
         self.latex_dir = self.code_dir / self.config.get("latex_dir", "Latex")
         self.figures_dir = self.code_dir / self.config.get("figures_dir", "Latex/figures")
 
-        # State file paths
+        # State file paths (agent working state — stays as YAML)
         self.state_file = self.state_dir / "research_state.yaml"
         self.findings_file = self.state_dir / "findings.yaml"
         self.paper_state_file = self.state_dir / "paper_state.yaml"
@@ -221,6 +226,24 @@ def display_name(self) -> str:
             self._display_name = title or name or self.project_name
         return self._display_name
 
+    # ========== DB Sync ==========
+
+    def _sync_db(self, **kwargs):
+        """Update project record in the webapp DB. Fail-soft: errors are logged, never raised."""
+        if not self._db_path or not self._project_id:
+            return
+        try:
+            from ark.webapp.db import get_session, get_project, update_project
+            with get_session(self._db_path) as session:
+                project = get_project(session, self._project_id)
+                if project:
+                    update_project(session, project, **kwargs)
+            self._db_sync_errors = 0
+        except Exception as e:
+            self._db_sync_errors += 1
+            if self._db_sync_errors <= 3:
+                self.log(f"DB sync failed ({self._db_sync_errors}): {e}", "WARN")
+
     # ========== Deep Research (background) ==========
 
     def _start_deep_research_background(self):
@@ -636,6 +659,7 @@ def set_language_pref(self, lang: str):
             data["language"] = lang
             with open(prefs_file, "w") as f:
                 yaml.dump(data, f, default_flow_style=False)
+            self._sync_db(language=lang)
             self.log(f"Language preference set to: {lang}", "INFO")
         except Exception as e:
             self.log(f"Failed to save language pref: {e}", "WARN")
@@ -947,6 +971,12 @@ def save_checkpoint(self):
         with open(self.checkpoint_file, "w") as f:
             yaml.dump(checkpoint, f, default_flow_style=False)
         self.log(f"Checkpoint saved: iteration={self.iteration}", "INFO")
+        self._sync_db(
+            checkpoint_data=json.dumps(checkpoint),
+            iteration=self.iteration,
+            total_input_tokens=self.total_input_tokens,
+            total_output_tokens=self.total_output_tokens,
+        )
 
     def save_step_checkpoint(self, step_num: int, step_name: str):
         """Save checkpoint after a step completes within a phase iteration."""
@@ -965,6 +995,7 @@ def save_step_checkpoint(self, step_num: int, step_name: str):
         }
         with open(self.checkpoint_file, "w") as f:
             yaml.dump(checkpoint, f, default_flow_style=False)
+        self._sync_db(checkpoint_data=json.dumps(checkpoint))
 
     # Backward compat alias
     save_phase_checkpoint = save_step_checkpoint
@@ -1206,6 +1237,25 @@ def save_paper_state(self, state: dict):
         """Save paper review state."""
         with open(self.paper_state_file, "w") as f:
             yaml.dump(state, f, default_flow_style=False, allow_unicode=True)
+        # Sync to DB
+        db_update = {
+            "score": float(state.get("current_score", 0)),
+            "iteration": self.iteration,
+        }
+        reviews = state.get("reviews", [])
+        if reviews:
+            db_update["score_history"] = json.dumps([
+                {"iteration": r.get("iteration", i + 1),
+                 "score": float(r.get("score", 0)),
+                 "timestamp": r.get("timestamp", "")}
+                for i, r in enumerate(reviews)
+            ])
+        paper_status = state.get("status", "in_progress")
+        if paper_status in ("accepted", "accepted_pending_cleanup"):
+            db_update["phase"] = "accepted"
+        else:
+            db_update["phase"] = "review"
+        self._sync_db(**db_update)
 
     def load_paper_requirements(self) -> dict:
         """Load paper requirements config."""
@@ -2046,8 +2096,19 @@ def main():
                         help="Override code directory (default: from project config)")
     parser.add_argument("--project-dir", type=str, default=None,
                         help="Override project directory (default: ARK_ROOT/projects/<project>)")
+    parser.add_argument("--db-path", type=str, default=None,
+                        help="Path to webapp SQLite DB for status sync")
+    parser.add_argument("--project-id", type=str, default=None,
+                        help="Project UUID in the webapp DB")
     args = parser.parse_args()
 
+    # Resolve DB path: explicit arg > env > webapp.env > default
+    db_path = args.db_path
+    project_id = args.project_id
+    if not db_path:
+        from ark.webapp.db import resolve_db_path
+        db_path = resolve_db_path()
+
     # Load project config to resolve code_dir if not specified
     project_dir = args.project_dir
     config_file = (Path(project_dir) if project_dir else ARK_ROOT / "projects" / args.project) / "config.yaml"
@@ -2058,6 +2119,21 @@ def main():
             cfg = _yaml.safe_load(f) or {}
         code_dir = cfg.get("code_dir")
 
+    # Auto-resolve project_id from DB if not provided
+    if not project_id and db_path and Path(db_path).exists():
+        try:
+            from ark.webapp.db import get_session, get_project_by_name, get_project
+            with get_session(db_path) as session:
+                # Try looking up by project name or by project_dir matching id
+                p = get_project_by_name(session, args.project)
+                if not p:
+                    # Maybe --project is actually a UUID
+                    p = get_project(session, args.project)
+                if p:
+                    project_id = p.id
+        except Exception:
+            pass
+
     orchestrator = Orchestrator(
         max_days=args.max_days,
         max_iterations=args.iterations,
@@ -2066,8 +2142,28 @@ def main():
         model=args.model,
         code_dir=code_dir,
         project_dir=project_dir,
+        db_path=db_path,
+        project_id=project_id,
     )
-    orchestrator.run()
+
+    # Mark as running in DB
+    if db_path and project_id:
+        orchestrator._sync_db(status="running", pid=os.getpid())
+
+    try:
+        orchestrator.run()
+        # Mark completion in DB
+        if db_path and project_id:
+            paper_state = orchestrator.load_paper_state()
+            final_status = "done"
+            if paper_state.get("status") in ("accepted", "accepted_pending_cleanup"):
+                final_status = "done"
+            orchestrator._sync_db(status=final_status, pid=0)
+    except KeyboardInterrupt:
+        orchestrator._sync_db(status="stopped", pid=0)
+    except Exception:
+        orchestrator._sync_db(status="failed", pid=0)
+        raise
 
 
 if __name__ == "__main__":

ark/pipeline.py

@@ -808,6 +808,7 @@ def _run_research_phase(self):
             4.1 clone conda env (ark-base → .env/)
             4.2 bootstrap citations → references.bib
         """
+        self._sync_db(phase="research")
         self.log("", "RAW")
         self.log_section("Research Phase  |  Understanding Project & Building Foundation")
 
@@ -1257,6 +1258,7 @@ def _update_title_from_context(self):
                     cfg["title"] = title
                     with open(config_file, "w") as f:
                         yaml.dump(cfg, f, default_flow_style=False, allow_unicode=True, sort_keys=False)
+                    self._sync_db(title=title, name=title)
                     self.log(f"Title updated: {title}", "INFO")
 
     # ==================== Citation Bootstrapping ====================
@@ -1308,6 +1310,7 @@ def _generate_title_if_needed(self):
                     cfg = yaml.safe_load(config_path.read_text()) or {}
                     cfg["title"] = new_title
                     config_path.write_text(yaml.dump(cfg, default_flow_style=False, allow_unicode=True))
+                self._sync_db(title=new_title, name=new_title)
 
     def _bootstrap_citations_from_deep_research(self):
         """Extract paper titles from Deep Research report via LLM, then fetch BibTeX via API.
@@ -1544,6 +1547,14 @@ def _save_dev_phase_state(self, state: dict):
         dev_state_file = self.state_dir / "dev_phase_state.yaml"
         with open(dev_state_file, "w") as f:
             yaml.dump(state, f, default_flow_style=False, allow_unicode=True)
+        # Sync to DB
+        dev_status = state.get("status", "pending")
+        phase = "dev" if dev_status == "in_progress" else ("review" if dev_status in ("completed", "complete") else "")
+        self._sync_db(
+            dev_iteration=int(state.get("iteration", 0)),
+            dev_status=dev_status,
+            phase=phase,
+        )
 
     def _run_dev_phase(self):
         """Run the Dev Phase: iterative experiments → initial paper draft.
@@ -2648,6 +2659,13 @@ def _write_cost_report(self):
             except Exception:
                 pass
             raise
+        # Sync cost totals to DB
+        self._sync_db(
+            total_cost_usd=round(total_cost_usd, 6),
+            total_input_tokens=total_input_tokens,
+            total_output_tokens=total_output_tokens,
+            total_agent_calls=total_calls,
+        )
 
     def run(self):
         """Main loop."""

ark/webapp/app.py

@@ -443,6 +443,15 @@ async def lifespan(app: FastAPI):
         pass
     # Now create engine + tables (ORM will see the migrated schema)
     get_engine(settings.db_path)
+
+    # Migrate existing project data: populate new DB columns from YAML state files
+    from ark.webapp.db import migrate_project_data
+    try:
+        migrate_project_data(settings.db_path, str(settings.projects_root))
+        logger.info("Project data migration completed.")
+    except Exception as e:
+        logger.warning(f"Project data migration failed (non-fatal): {e}")
+
     logger.info(f"ARK Webapp starting. DB: {settings.db_path}")
     logger.info(f"Projects root: {settings.projects_root}")