348. 348 /// lifecycle.
349. 349 fn evict_idle_sessions(&self) {
350. 350 let cap = max_live_sessions();
351. 351 let mut sessions = self
352. 352 .sessions
353. 353 .lock()
354. 354 .unwrap_or_else(std::sync::PoisonError::into_inner);
355. 355 if sessions.len() <= cap {
356. 356 return;
357. 357 }
358. 358 let mut idle: Vec<(std::time::Instant, String)> = sessions
359. 359 .iter()
360. 360 .filter(|(_, handle)| {
361. 361 // `events_tx` always carries the handle's own internal
362. 362 // projector (`register_handle`), so `receiver_count()` can
363. 363 // never read zero; `external_subscribers()` excludes it.
364. 364 // `side_events_tx` has no such internal subscriber, so a
365. 365 // plain `receiver_count()` remains correct there.
366. 366 Arc::strong_count(handle) == 1
367. 367 && handle.events_tx.external_subscribers() == 0
368. 368 && handle.side_events_tx.receiver_count() == 0
369. 369 && handle.session.lock().is_ok_and(|guard| guard.is_some())
370. 370 })
371. 371 .filter_map(|(id, handle)| {
372. 372 let touched = *handle.last_touched.lock().ok()?;
373. 373 Some((touched, id.clone()))
374. 374 })
375. 375 .collect();
376. 376 idle.sort_by_key(|(touched, _)| *touched);
377. 377 let mut over = sessions.len().saturating_sub(cap);
378. 378 for (_, id) in idle {
379. 379 if over == 0 {
380. 380 break;
381. 381 }
382. 382 sessions.remove(&id);
383. 383 over -= 1;
384. 384 }
385. 385 }
386. 386
387. 387 /// Drops a session's live handle, so a trashed session is not served
388. 388 /// from memory after it leaves every list.
389. 389 fn forget_session(&self, id: &str) {
390. 390 self.sessions
391. 391 .lock()
392. 392 .unwrap_or_else(std::sync::PoisonError::into_inner)
393. 393 .remove(id);
394. 394 }
395. 395
396. 396 /// Snapshot of every live session handle (admin surfaces aggregate
397. 397 /// across sessions; nothing here crosses a session's approval scope —
398. 398 /// answering still goes through the per-session endpoint).
399. 399 pub(crate) fn live_handles(&self) -> Vec<Arc<SessionHandle>> {
400. 400 self.sessions
401. 401 .lock()
402. 402 .unwrap_or_else(std::sync::PoisonError::into_inner)
403. 403 .values()
404. 404 .cloned()
405. 405 .collect()
406. 406 }
407. 407}
408. 408
409. 409#[cfg(test)]
410. 410#[allow(clippy::unwrap_used, clippy::expect_used)]
411. 411mod eviction_tests {
412. 412 use super::*;
413. 413
414. 414 /// Restores the process-global test override on drop, so a panic mid-test
415. 415 /// cannot leave a tiny cap active for an unrelated concurrent test.
416. 416 struct RestoreCap;
417. 417 impl Drop for RestoreCap {
418. 418 fn drop(&mut self) {
419. 419 MAX_LIVE_SESSIONS_TEST_OVERRIDE.store(0, std::sync::atomic::Ordering::SeqCst);
420. 420 }
421. 421 }
422. 422
423. 423 /// Finding 3: every handle's `events_tx` carries the internal projector
424. 424 /// subscription from `register_handle`, so the old `receiver_count() ==
425. 425 /// 0` eviction guard never held — a long-lived process kept every
426. 426 /// session's ledger (and its exclusive file lock) in memory forever.
427. 427 /// `external_subscribers()` fixes the guard; this proves eviction
428. 428 /// actually runs once the live set exceeds the cap.
429. 429 #[tokio::test]
430. 430 async fn idle_sessions_beyond_the_cap_are_evicted() {
431. 431 vak_config::paths::isolate_home_for_tests();
432. 432 MAX_LIVE_SESSIONS_TEST_OVERRIDE.store(3, std::sync::atomic::Ordering::SeqCst);
433. 433 let _restore = RestoreCap;
434. 434
435. 435 let dir = tempfile::tempdir().unwrap();
436. 436 let core = Core::new(dir.path().to_path_buf()).unwrap();
437. 437 core.set_sessions_home(dir.path().join("home"));
438. 438 let state = AppState::new(core.clone());
439. 439
440. 440 let mut ids = Vec::new();
441. 441 for _ in 0..5 {
442. 442 let session = core.start_session().await.unwrap();
443. 443 let id = session.header().unwrap().session_id.clone();
444. 444 // No SSE client, no run in progress, and the returned Arc is
445. 445 // dropped immediately — exactly the "nothing references it"
446. 446 // shape `evict_idle_sessions` looks for.
447. 447 let _ = register_handle(
448. 448 &state,
449. 449 id.clone(),
450. 450 session,
451. 451 core.cwd().clone(),
452. 452 core.clone(),
453. 453 );
454. 454 ids.push(id);
455. 455 }
456. 456
457. 457 let live: Vec<String> = state.sessions.lock().unwrap().keys().cloned().collect();
458. 458 assert_eq!(
459. 459 live.len(),
460. 460 3,
461. 461 "expected eviction down to the test cap, got {live:?}"
462. 462 );
463. 463 // Eviction drops the least-recently-touched handles first, so the
464. 464 // most recently created session must survive.
465. 465 assert!(
466. 466 live.contains(ids.last().unwrap()),
467. 467 "the newest session must not be evicted: {live:?}"
468. 468 );
469. 469 assert!(
470. 470 !live.contains(&ids[0]),
471. 471 "the oldest session must be evicted first: {live:?}"
472. 472 );
473. 473 }
474. 474}
475. 475
476. 476#[derive(Clone)]
477. 477struct CoworkingPresence {
478. 478 display_name: String,
479. 479 seen_at: Instant,
480. 480 office_room_id: Option<String>,
481. 481 office_anchor: Option<String>,
482. 482}
483. 483
484. 484#[derive(Clone)]
485. 485pub struct ApprovalRequest {
486. 486 pub id: String,
487. 487 pub tool: String,
488. 488 pub args_json: String,
489. 489 pub reason: String,
490. 490 pub requested_at: chrono::DateTime<chrono::Utc>,
491. 491 respond: Arc<Mutex<Option<oneshot::Sender<bool>>>>,
492. 492 answered_by: Arc<Mutex<Option<(String, String)>>>,
493. 493 delegated_to: Arc<Mutex<Option<String>>>,
494. 494}
495. 495
496. 496impl ApprovalRequest {
497. 497 pub fn respond(&self, approve: bool) {
498. 498 if let Some(tx) = self
499. 499 .respond
500. 500 .lock()
501. 501 .unwrap_or_else(std::sync::PoisonError::into_inner)
502. 502 .take()
503. 503 {
504. 504 let _ = tx.send(approve);
505. 505 }
506. 506 }
507. 507}
508. 508
509. 509/// How long an HTTP-surfaced gate waits for a console or desktop client to
510. 510/// answer before failing closed.
511. 511///
512. 512/// There was no bound at all, which was survivable while every run behind
513. 513/// this approver had a human watching an SSE stream — and was not, once the
514. 514/// scheduler started firing runs through the same path. An unanswered gate
515. 515/// held the session handle open forever, so the routine never completed and
516. 516/// its slot never freed. Generous, because a person may genuinely be away
517. 517/// from the tab, but finite: a run that fails closed can be retried, and one
518. 518/// that hangs cannot.
519. 519const HTTP_APPROVAL_TIMEOUT: Duration = Duration::from_secs(900);
520. 520
521. 521struct HttpApprover {
522. 522 events_tx: events::EventBus,
523. 523 pending: Arc<Mutex<HashMap<String, ApprovalRequest>>>,
524. 524 /// Owning session, so admin-console surfaces can attribute gates.
525. 525 session_id: String,
526. 526 activity_buffer: Arc<Mutex<Vec<vak_session::ActivityRecord>>>,
527. 527 /// False when this run has no client watching — a scheduled routine, a
528. 528 /// best-of-N leg. The gate is then a foregone denial, and saying so
529. 529 /// through `answerable()` is what lets `vak_core::reach` drop the
530. 530 /// capability from the turn instead of letting the model discover it by
531. 531 /// blocking on a question nobody will read.
532. 532 answerable: bool,
533. 533}
534. 534
535. 535#[async_trait::async_trait]
536. 536impl Approver for HttpApprover {
537. 537 fn answerable(&self) -> bool {
538. 538 self.answerable
539. 539 }
540. 540
541. 541 async fn approve(&self, tool: &str, args_json: &str, reason: &str) -> bool {
542. 542 if !self.answerable {
543. 543 return false;
544. 544 }
545. 545 let id = uuid::Uuid::now_v7().to_string();
546. 546 let (respond, rx) = oneshot::channel();
547. 547 let answered_by = Arc::new(Mutex::new(None));
548. 548 self.pending
549. 549 .lock()
550. 550 .unwrap_or_else(std::sync::PoisonError::into_inner)
551. 551 .insert(
552. 552 id.clone(),
553. 553 ApprovalRequest {
554. 554 id: id.clone(),
555. 555 tool: tool.to_string(),
556. 556 args_json: args_json.to_string(),
557. 557 reason: reason.to_string(),
558. 558 requested_at: chrono::Utc::now(),
559. 559 respond: Arc::new(Mutex::new(Some(respond))),
560. 560 answered_by: answered_by.clone(),
561. 561 delegated_to: Arc::new(Mutex::new(None)),
562. 562 },
563. 563 );
564. 564 let _ = self.events_tx.send(AgentEvent::ApprovalRequested {
565. 565 id: id.clone(),
566. 566 tool: tool.to_string(),
567. 567 args_json: args_json.to_string(),
568. 568 reason: reason.to_string(),
569. 569 });
570. 570 self.activity_buffer
571. 571 .lock()
572. 572 .unwrap_or_else(std::sync::PoisonError::into_inner)
573. 573 .push(vak_session::ActivityRecord {
574. 574 activity_id: format!("approval-{id}"),
575. 575 turn: None,
576. 576 kind: vak_session::ActivityKind::Approval,
577. 577 status: vak_session::ActivityStatus::Pending,
578. 578 label: format!("Approval required for {tool}"),
579. 579 detail: Some(reason.to_string()),
580. 580 data: [
581. 581 ("request_id".into(), id.clone()),
582. 582 ("tool".into(), tool.to_string()),
583. 583 ("args_json".into(), args_json.to_string()),
584. 584 ]
585. 585 .into(),
586. 586 });
587. 587 if let Some(hub) = events::global() {
588. 588 hub.emit(events::SystemEvent::ApprovalRequested {
589. 589 id: id.clone(),
590. 590 session_id: self.session_id.clone(),
591. 591 tool: tool.to_string(),
592. 592 reason: reason.to_string(),
593. 593 });
594. 594 }
595. 595 // Bounded, and failing closed on expiry — the same contract
596. 596 // `GatewayApprover` already had. Dropping the entry before returning
597. 597 // means a reply that arrives after the deadline resolves nothing
598. 598 // rather than answering a gate the run has already moved past.
599. 599 let approved = match tokio::time::timeout(HTTP_APPROVAL_TIMEOUT, rx).await {
600. 600 Ok(answer) => answer.unwrap_or(false),
601. 601 Err(_) => {
602. 602 eprintln!(
603. 603 "[approvals] gate {} for `{tool}` expired after {}s; denied",
604. 604 &id[..8.min(id.len())],
605. 605 HTTP_APPROVAL_TIMEOUT.as_secs()
606. 606 );
607. 607 false
608. 608 }
609. 609 };
610. 610 self.pending
611. 611 .lock()
612. 612 .unwrap_or_else(std::sync::PoisonError::into_inner)
613. 613 .remove(&id);
614. 614 let answered_by = answered_by
615. 615 .lock()
616. 616 .unwrap_or_else(std::sync::PoisonError::into_inner)
617. 617 .clone();
618. 618 self.activity_buffer
619. 619 .lock()
620. 620 .unwrap_or_else(std::sync::PoisonError::into_inner)
621. 621 .push(vak_session::ActivityRecord {
622. 622 activity_id: format!("approval-{id}"),
623. 623 turn: None,
624. 624 kind: vak_session::ActivityKind::Approval,
625. 625 status: if approved {
626. 626 vak_session::ActivityStatus::Succeeded
627. 627 } else {
628. 628 vak_session::ActivityStatus::Denied
629. 629 },
630. 630 label: format!(
631. 631 "Approval {} for {tool}",
632. 632 if approved { "granted" } else { "denied" }
633. 633 ),
634. 634 detail: Some(reason.to_string()),
635. 635 data: [
636. 636 ("request_id".into(), id.clone()),
637. 637 ("tool".into(), tool.to_string()),
638. 638 ("args_json".into(), args_json.to_string()),
639. 639 ]
640. 640 .into_iter()
641. 641 .chain(answered_by.into_iter().flat_map(|(actor_id, actor_name)| {
642. 642 [
643. 643 ("actor_id".into(), actor_id),
644. 644 ("actor_name".into(), actor_name),
645. 645 ]
646. 646 }))
647. 647 .collect(),
648. 648 });
649. 649 if let Some(hub) = events::global() {
650. 650 hub.emit(if approved {
651. 651 events::SystemEvent::ApprovalGranted {
652. 652 id: id.clone(),
653. 653 tool: tool.to_string(),
654. 654 }
655. 655 } else {
656. 656 events::SystemEvent::ApprovalDenied {
657. 657 id,
658. 658 tool: tool.to_string(),
659. 659 }
660. 660 });
661. 661 }
662. 662 approved
663. 663 }
664. 664}
665. 665
666. 666/// Resolve the Agent-scoped `Core` for the rest of an async handler body, or
667. 667/// return early with `resolve_scoped_core`'s own error `Response` (forwarding
668. 668/// its status/message unchanged, e.g. 409 CONFLICT for a paused/archived
669. 669/// Agent). Every endpoint scoped this way needs the identical
670. 670/// match-and-early-return, so it lives here once instead of copy-pasted at
671. 671/// each of the 50+ call sites (see `resolve_scoped_core` below).
672. 672macro_rules! scoped_core {
673. 673 ($state:expr, $session_id:expr, $agent:expr) => {
674. 674 match resolve_scoped_core($state, $session_id, $agent) {
675. 675 Ok(core) => core,
676. 676 Err(response) => return response,
677. 677 }
678. 678 };
679. 679}
680. 680
681. 681pub fn router(core: Core) -> Router {
682. 682 router_with_state(AppState::new(core))
683. 683}
684. 684
685. 685/// Unauthenticated router with the gateway force-enabled and no background
686. 686/// scheduler. For embedders that run their own supervision loop and need
687. 687/// clean teardown: dropping this router releases every session lock,
688. 688/// whereas `secured_router`'s scheduler pins handles until process exit.
689. 689pub fn gateway_router(core: Core) -> Router {
690. 690 let mut state = AppState::new(core);
691. 691 state.enable_gateway();
692. 692 router_with_state(state)
693. 693}
694. 694
695. 695fn router_with_state(state: AppState) -> Router {
696. 696 Router::new()
697. 697 .route("/health", get(health))
698. 698 .route("/sessions", get(list_sessions).post(create_session))
699. 699 .route("/sessions/{id}/attach", post(attach_session))
700. 700 .route("/sessions/{id}/diff", get(session_diff))
701. 701 .route("/sessions/{id}/receipts", get(session_receipts))
702. 702 .route(
703. 703 "/sessions/{id}/work",
704. 704 get(session_work).post(session_work_command),
705. 705 )
706. 706 .route("/sessions/{id}/work/confirm", post(session_work_confirm))
707. 707 .route("/sessions/{id}/work/revise", post(session_work_revise))
708. 708 .route(
709. 709 "/sessions/{id}/work/items/{item_id}/retry",
710. 710 post(session_work_retry),
711. 711 )
712. 712 .route(
713. 713 "/sessions/{id}/work/items/{item_id}/cancel",
714. 714 post(session_work_cancel_item),
715. 715 )
716. 716 .route(
717. 717 "/sessions/{id}/work/items/{item_id}/reassign",
718. 718 post(session_work_reassign),
719. 719 )
720. 720 .route("/flows", get(flows_list))
721. 721 .route("/flows/{name}/runs", get(flow_runs_list))
722. 722 .route("/flows/{name}/runs/{run}/graph", get(flow_run_graph))
723. 723 .route("/sessions/{id}/checkpoints", get(list_checkpoints))
724. 724 .route(
725. 725 "/sessions/{id}/checkpoints/{seq}/restore",
726. 726 post(restore_checkpoint),
727. 727 )
728. 728 .route("/sessions/{id}/archive", post(set_archived))
729. 729 .route("/sessions/archived", delete(delete_all_archived))
730. 730 .route("/sessions/{id}", delete(delete_session))
731. 731 .route("/sessions/{id}/restore", post(restore_session))
732. 732 .route("/skills", get(list_skills))
733. 733 .route("/commands", get(list_commands))
734. 734 .route("/plugins", get(list_plugins))
735. 735 .route("/plugins/catalog", get(plugin_catalog))
736. 736 .route("/plugins/audit", get(plugin_audit))
737. 737 .route("/plugins/invocations", get(plugin_invocations))
738. 738 .route(
739. 739 "/presentations",
740. 740 get(list_presentations).post(register_presentations),
741. 741 )
742. 742 .route(
743. 743 "/presentations/specs/{id}/{revision}",
744. 744 get(get_presentation_spec),
745. 745 )
746. 746 .route(
747. 747 "/presentations/revisions",
748. 748 post(propose_presentation_revision),
749. 749 )
750. 750 .route(
751. 751 "/sessions/{id}/presentation/proposals",
752. 752 post(propose_session_presentation_revision),
753. 753 )
754. 754 .route("/presentations/export", get(export_presentations))
755. 755 .route("/presentations/import", post(import_presentations))
756. 756 .route(
757. 757 "/presentations/activate-all",
758. 758 post(activate_all_presentations),
759. 759 )
760. 760 .route(
761. 761 "/presentations/deactivate-all",
762. 762 post(deactivate_all_presentations),
763. 763 )
764. 764 .route(
765. 765 "/presentations/{id}/{revision}/activate",
766. 766 post(activate_presentation),
767. 767 )
768. 768 .route(
769. 769 "/presentations/{id}/deactivate",
770. 770 post(deactivate_presentation),
771. 771 )
772. 772 .route("/presentations/{id}/reset", post(reset_presentation))
773. 773 .route(
774. 774 "/presentations/plugins/{plugin_id}",
775. 775 delete(revoke_presentations_plugin),
776. 776 )
777. 777 .route(
778. 778 "/plugins/retired",
779. 779 get(list_retired_plugins).delete(remove_retired_plugins),
780. 780 )
781. 781 .route(
782. 782 "/plugins/sources",
783. 783 get(plugin_sources).post(plugin_register_source),
784. 784 )
785. 785 .route("/plugins/sources/{id}/enable", post(plugin_source_enable))
786. 786 .route("/plugins/sources/{id}/disable", post(plugin_source_disable))
787. 787 .route("/plugins/keys/{id}/revoke", post(plugin_key_revoke))
788. 788 .route("/plugins/keys/{id}/restore", post(plugin_key_restore))
789. 789 .route("/plugins/install", post(plugin_install))
790. 790 .route("/plugins/update", post(plugin_update))
791. 791 .route("/plugins/{name}/enable", post(plugin_enable))
792. 792 .route("/plugins/{name}/disable", post(plugin_disable))
793. 793 .route("/plugins/{name}/rollback", post(plugin_rollback))
794. 794 .route("/plugins/{name}", delete(plugin_remove))
795. 795 .route("/sessions/{id}/pr", get(session_pr))
796. 796 .route("/sessions/{id}/pr/merge", post(pr_merge))
797. 797 .route("/tasks", get(list_tasks).post(create_task))
798. 798 .route(
799. 799 "/tasks/{id}",
800. 800 axum::routing::patch(patch_task).delete(delete_task),
801. 801 )
802. 802 .route("/tasks/{id}/run-now", post(run_task_now))
803. 803 .route("/tasks/{id}/retry-delivery", post(retry_task_delivery))
804. 804 .route("/sessions/{id}/launch", get(get_launch))
805. 805 .route("/sessions/{id}/launch/prepare", post(prepare_launch))
806. 806 .route("/sessions/{id}/launch/start", post(start_launch))
807. 807 .route("/sessions/{id}/launch/stop", post(stop_launch))
808. 808 .route("/sessions/{id}/launch/logs", get(launch_logs))
809. 809 .route("/sessions/{id}/run", post(run_prompt))
810. 810 .route("/sessions/{id}/steering", post(send_steering))
811. 811 .route("/sessions/{id}/cancel", post(cancel_run))
812. 812 .route("/sessions/{id}/pause", post(pause_run))
813. 813 .route("/sessions/{id}/resume", post(resume_run))
814. 814 .route("/sessions/{id}/control-state", get(control_state))
815. 815 .route("/sessions/{id}/plan-change", post(plan_change))
816. 816 .route("/sessions/{id}/workers", get(list_workers))
817. 817 .route("/sessions/{id}/workers/{child}/steer", post(steer_worker))
818. 818 .route("/sessions/{id}/workers/{child}/stop", post(stop_worker))
819. 819 // Backward-compatible aliases for the old `subagents` route names.
820. 820 .route("/sessions/{id}/subagents", get(list_workers))
821. 821 .route("/sessions/{id}/subagents/{child}/steer", post(steer_worker))
822. 822 .route("/sessions/{id}/subagents/{child}/stop", post(stop_worker))
823. 823 .route("/sessions/{id}/approvals/{req_id}", post(answer_approval))
824. 824 .route("/sessions/{id}/outcome-review", post(record_outcome_review))
825. 825 .route("/sessions/{id}/events", get(events_sse))
826. 826 .route(
827. 827 "/sessions/{id}/sandbox/executions",
828. 828 get(session_sandbox_executions),
829. 829 )
830. 830 .route(
831. 831 "/sessions/{id}/sandbox/records",
832. 832 get(list_session_sandbox_records),
833. 833 )
834. 834 .route(
835. 835 "/sessions/{id}/sandbox/candidates",
836. 836 post(export_sandbox_candidate),
837. 837 )
838. 838 .route(
839. 839 "/sessions/{id}/sandbox/candidates/{candidate_id}/files",
840. 840 get(read_sandbox_candidate_file),
841. 841 )
842. 842 .route(
843. 843 "/sessions/{id}/sandbox/candidates/{candidate_id}/files/raw",
844. 844 get(read_sandbox_candidate_file_raw),
845. 845 )
846. 846 .route(
847. 847 "/sessions/{id}/sandbox/candidates/{candidate_id}/office-review",
848. 848 get(read_sandbox_candidate_office_review),
849. 849 )
850. 850 .route(
851. 851 "/sessions/{id}/sandbox/candidates/{candidate_id}/office-narrow",
852. 852 post(narrow_sandbox_candidate_office),
853. 853 )
854. 854 .route(
855. 855 "/sessions/{id}/sandbox/candidates/{candidate_id}/office",
856. 856 get(read_sandbox_candidate_office_projection),
857. 857 )
858. 858 .route(
859. 859 "/sessions/{id}/sandbox/candidates/{candidate_id}/comments",
860. 860 get(list_sandbox_candidate_comments).post(comment_on_sandbox_candidate),
861. 861 )
862. 862 .route(
863. 863 "/sessions/{id}/sandbox/candidates/{candidate_id}/comments/{comment_id}/request-revision",
864. 864 post(request_revision_from_candidate_comment),
865. 865 )
866. 866 .route(
867. 867 "/sessions/{id}/sandbox/promote",
868. 868 post(promote_sandbox_candidate),
869. 869 )
870. 870 .route(
871. 871 "/sessions/{id}/sandbox/promotions/{candidate_id}/undo",
872. 872 post(undo_sandbox_promotion),
873. 873 )
874. 874 .route(
875. 875 "/sessions/{id}/sandbox/promotions/{candidate_id}/checks",
876. 876 post(run_sandbox_workspace_check),
877. 877 )
878. 878 .route("/sessions/{id}/presentation", get(presentation_snapshot))
879. 879 .route("/sessions/{id}/results/{result_id}", get(session_result))
880. 880 .route(
881. 881 "/sessions/{id}/presentation/feedback",
882. 882 post(presentation_feedback),
883. 883 )
884. 884 .route(
885. 885 "/sessions/{id}/presentation/select",
886. 886 post(select_presentation_for_session),
887. 887 )
888. 888 .route(
889. 889 "/sessions/{id}/presentation/events",
890. 890 get(presentation_events_sse),
891. 891 )
892. 892 .route("/sessions/{id}/transcript", get(transcript))
893. 893 .route("/sessions/{id}/transcript.md", get(transcript_markdown))
894. 894 .route(
895. 895 "/sessions/{id}/coworking/invitations",
896. 896 get(list_coworking_invitations).post(create_coworking_invitation),
897. 897 )
898. 898 .route("/sessions/{id}/coworking/me", get(coworking_me))
899. 899 .route(
900. 900 "/sessions/{id}/coworking/presence",
901. 901 get(coworking_presence),
902. 902 )
903. 903 .route(
904. 904 "/sessions/{id}/coworking/messages",
905. 905 post(create_coworking_message),
906. 906 )
907. 907 .route(
908. 908 "/sessions/{id}/coworking/approvals",
909. 909 get(list_coworking_approvals),
910. 910 )
911. 911 .route(
912. 912 "/sessions/{id}/coworking/approvals/{req_id}",
913. 913 post(answer_coworking_approval),
914. 914 )
915. 915 .route(
916. 916 "/sessions/{id}/coworking/approvals/{req_id}/delegate",
917. 917 post(delegate_coworking_approval),
918. 918 )
919. 919 .route("/sessions/{id}/coworking/updates", get(coworking_updates))
920. 920 .route("/sessions/{id}/office-workspaces", get(office_workspace::list).post(office_workspace::create))
921. 921 .route("/sessions/{id}/office-workspaces/{room_id}", post(office_workspace::mutate))
922. 922 .route("/sessions/{id}/office-workspaces/{room_id}/presence", post(office_workspace::focus))
923. 923 .route(
924. 924 "/sessions/{id}/coworking/invitations/{grant_id}/revoke",
925. 925 post(revoke_coworking_invitation),
926. 926 )
927. 927 .route("/sessions/{id}/side", post(side_chat))
928. 928 .route("/sessions/{id}/side/cancel", post(side_cancel_run))
929. 929 .route("/sessions/{id}/bestofn", post(start_bestofn))
930. 930 .route("/sessions/{id}/keep", post(keep_best_run))
931. 931 .route("/sessions/{id}/discard", post(discard_best_run))
932. 932 .route("/fs/file", get(read_file).put(write_file))
933. 933 .route("/fs/file/raw", get(read_file_raw))
934. 934 .route("/fs/office", get(read_office_projection))
935. 935 .route(
936. 936 "/fs/inbox",
937. 937 post(upload_to_inbox).layer(axum::extract::DefaultBodyLimit::max(
938. 938 inbox::UPLOAD_MAX_BYTES,
939. 939 )),
940. 940 )
941. 941 .route("/fs/preview/{*path}", get(preview_file))
942. 942 .route("/sandbox/records", get(list_sandbox_records))
943. 943 .route("/fs/tree", get(fs_tree))
944. 944 .route("/config", get(get_config).patch(patch_config))
945. 945 .route("/config/intent/evidence", post(patch_evidence_policy))
946. 946 .route(
947. 947 "/config/global",
948. 948 get(get_global_config_layer).patch(patch_global_config),
949. 949 )
950. 950 .route("/config/workspace", get(get_workspace_config_layer))
951. 951 .route("/config/project", get(get_workspace_config_layer))
952. 952 .route("/config/mode", post(set_permission_mode))
953. 953 .route(
954. 954 "/agent-network/capabilities",
955. 955 post(agent_network_capability),
956. 956 )
957. 957 .route(
958. 958 "/agent-network/messages",
959. 959 post(agent_network_send).get(agent_network_receive),
960. 960 )
961. 961 .route("/config/mcp", get(get_mcp_servers).put(put_mcp_servers))
962. 962 .route(
963. 963 "/config/mcp/global",
964. 964 get(get_global_mcp_servers).put(put_global_mcp_servers),
965. 965 )
966. 966 .route("/config/integrations", get(get_integration_catalog))
967. 967 .route(
968. 968 "/config/integrations/{id}",
969. 969 get(get_scoped_integration)
970. 970 .put(put_scoped_integration)
971. 971 .delete(delete_scoped_integration),
972. 972 )
973. 973 .route("/config/hooks", get(get_hooks).put(put_hooks))
974. 974 .route(
975. 975 "/config/prompts",
976. 976 get(get_prompt_layer).put(put_prompt_block),
977. 977 )
978. 978 .route("/config/prompts/effective", get(get_prompt_effective))
979. 979 .route("/config/prompts/preview", post(preview_prompt))
980. 980 .route("/config/prompts/roles", get(list_prompt_roles))
981. 981 .route("/agents", get(agent_chats::list))
982. 982 .route("/agents/{agent}/open", post(agent_chats::open))
983. 983 .route("/config/agents", get(get_agents).put(put_agents))
984. 984 .route(
985. 985 "/config/hooks/global",
986. 986 get(get_global_hooks).put(put_global_hooks),
987. 987 )
988. 988 .route(
989. 989 "/config/key",
990. 990 put(put_provider_key).delete(delete_provider_key),
991. 991 )
992. 992 // Distributed event bus status (vak-bus, docs/design/53).
993. 993 // Credentials are never returned; only the connection state and
994. 994 // metrics are exposed.
995. 995 .route(
996. 996 "/config/bus",
997. 997 get(get_bus_config)
998. 998 .put(put_bus_config)
999. 999 .delete(delete_bus_config),
1000. 1000 )
1001. 1001 // The approval policy: whether an `Ask` raised on an unattended
1002. 1002 // chat surface reaches a human at all. Read-only everywhere until
1003. 1003 // now, which made `vak_core::reach`'s own printed remedy an action
1004. 1004 // no surface could perform.
1005. 1005 .route(
1006. 1006 "/gateway/approvals",
1007. 1007 get(get_gateway_approvals).put(put_gateway_approvals),
1008. 1008 )
1009. 1009 // The three permission rule lists, as the engine evaluates them.
1010. 1010 .route(
1011. 1011 "/config/permissions",
1012. 1012 get(get_permission_rules).put(put_permission_rules),
1013. 1013 )
1014. 1014 // Multi-bot-per-surface (docs/design/34, multi-bot): a `Bot` is an
1015. 1015 // independent identity, so it gets its own id-addressed routes
1016. 1016 // rather than reusing the one-slot-per-surface ones above.
1017. 1017 .route("/gateway/bots", get(list_bots).post(create_bot))
1018. 1018 .route(
1019. 1019 "/gateway/bots/{id}",
1020. 1020 axum::routing::patch(update_bot).delete(delete_bot),
1021. 1021 )
1022. 1022 .route(
1023. 1023 "/gateway/bots/{id}/token",
1024. 1024 put(put_bot_id_token).delete(delete_bot_id_token),
1025. 1025 )
1026. 1026 .route("/providers", get(list_providers))
1027. 1027 .route("/providers/{name}/models", get(discover_models))
1028. 1028 .route(
1029. 1029 "/providers/{name}/models/availability",
1030. 1030 get(model_availability),
1031. 1031 )
1032. 1032 .route("/providers/{name}/status", get(provider_status))
1033. 1033 .route("/search", get(search_sessions))
1034. 1034 .route("/ops/status", get(ops_status))
1035. 1035 .route("/ops/center", get(operations_center))
1036. 1036 .route("/ops/actions", get(operations_actions))
1037. 1037 .route("/ops/incidents", get(operations_incidents))
1038. 1038 .route("/ops/outbox", get(operations_outbox))
1039. 1039 .route(
1040. 1040 "/ops/outbox/{job_id}/replay",
1041. 1041 post(replay_operations_outbox),
1042. 1042 )
1043. 1043 .route("/ops/{service}/{action}", post(ops_action))
1044. 1044 .route("/ops/diagnostics", get(ops_diagnostics))
1045. 1045 // Activation, explicitly (docs/design/46 D6). Configuration writes
1046. 1046 // never register a service; this is the one action that does.
1047. 1047 .route("/ops/services/activate", post(activate_services))
1048. 1048 .route("/finops", get(finops_status).patch(patch_finops))
1049. 1049 .route("/voice/speak", post(voice::voice_speak))
1050. 1050 .route("/voice/providers", get(voice::voice_providers))
1051. 1051 .route("/memory", get(list_memory).post(append_memory))
1052. 1052 .route("/memory/cleanup", post(cleanup_memory))
1053. 1053 .route("/memory/consolidate", post(consolidate_memory_route))
1054. 1054 .route(
1055. 1055 "/memory/{note_id}",
1056. 1056 axum::routing::patch(amend_memory_note).delete(forget_memory_note),
1057. 1057 )
1058. 1058 .route(
1059. 1059 "/entities",
1060. 1060 get(list_entities_route).post(upsert_entity_route),
1061. 1061 )
1062. 1062 .route(
1063. 1063 "/entities/{id}",
1064. 1064 get(get_entity_route).delete(delete_entity_route),
1065. 1065 )
1066. 1066 .route("/agents/templates", get(list_agent_templates))
1067. 1067 .route("/agents/instantiate", post(instantiate_agent_template))
1068. 1068 .route("/canvas/preview", post(canvas_preview))
1069. 1069 .route("/intent/explain", get(intent_explain))
1070. 1070 .route("/intent/policy", get(intent_policy))
1071. 1071 .route("/commitments", get(list_commitments))
1072. 1072 .route("/commitments/{id}", get(get_commitment))
1073. 1073 .route("/commitments/{id}/close", post(close_commitment))
1074. 1074 .route("/doctor", get(doctor_report))
1075. 1075 .route("/onboarding", get(onboarding_state))
1076. 1076 // The composition layer setup needs, and nothing more: every other
1077. 1077 // choice reuses the config, provider, integration, and bot APIs
1078. 1078 // that already exist (doc 46, "API and command design").
1079. 1079 .route("/onboarding/seed", post(onboarding_seed))
1080. 1080 .route(
1081. 1081 "/onboarding/workspace-review",
1082. 1082 post(onboarding_workspace_review),
1083. 1083 )
1084. 1084 .route("/onboarding/trust", post(onboarding_trust))
1085. 1085 .route("/onboarding/first-task", post(onboarding_first_task))
1086. 1086 // ---- browser auth (docs/design/48-web-client.md §4.3) -----------
1087. 1087 //
1088. 1088 // ONE login for every browser surface — the workspace client and
1089. 1089 // the operations console share this exchange and this cookie.
1090. 1090 // These replace `/admin/login` and `/admin/logout`, which were
1091. 1091 // removed rather than kept alongside: two endpoints against one
1092. 1092 // cookie is two contracts that must agree forever (invariant 30).
1093. 1093 .route("/auth/login", post(web::login))
1094. 1094 .route("/auth/logout", post(web::logout))
1095. 1095 .route("/auth/session", get(web::session_status))
1096. 1096 // ---- the workspace client's own host surface --------------------
1097. 1097 .route("/host", get(web::host_info))
1098. 1098 .route("/stream", get(stream::stream))
1099. 1099 .route("/workspaces", get(web::list_workspaces))
1100. 1100 .route("/workspaces/open", post(web::open_workspace))
1101. 1101 .route("/workspaces/forget", post(web::forget_workspace))
1102. 1102 .route("/fs/dirs", get(web::list_dirs))
1103. 1103 .route("/pty", get(web::pty_socket))
1104. 1104 .route("/voice/session", get(voice::voice_socket))
1105. 1105 .route("/version", get(web::version))
1106. 1106 .route("/backup/export", post(backup_export))
1107. 1107 .route("/backup/import", post(backup_import))
1108. 1108 .route("/digest", get(digest_report))
1109. 1109 .route("/inbox", get(inbox_list))
1110. 1110 .route("/inbox/unread_count", get(inbox_unread_count))
1111. 1111 .route("/inbox/{id}/ack", post(inbox_ack))
1112. 1112 .route("/skills/proposals", get(list_proposals_route))
1113. 1113 .route("/skills/proposals/{id}/promote", post(promote_proposal))
1114. 1114 .route("/skills/proposals/{id}/reject", post(reject_proposal))
1115. 1115 .merge(gateway::routes())
1116. 1116 .merge(feeds::routes())
1117. 1117 .merge(admin::routes())
1118. 1118 // Compression is scoped to the three STATIC bundles and nowhere
1119. 1119 // else. These are the big, highly compressible responses — a site
1120. 1120 // page is ~78 KB of inlined CSS and markup, the vendored motion
1121. 1121 // build is 141 KB, and the SPA bundles are larger still — and this
1122. 1122 // product is explicitly built to be reached over a tunnel, where
1123. 1123 // that is the whole first-visit cost.
1124. 1124 //
1125. 1125 // It is NOT applied to the API. `/sessions/:id/events` is
1126. 1126 // server-sent events: a compressor sits between the writer and the
1127. 1127 // socket, and a live transcript that arrives in buffer-sized
1128. 1128 // batches instead of per frame is a worse product than an
1129. 1129 // uncompressed one. Scoping it here rather than at the root is the
1130. 1130 // difference between a smaller page and a laggy agent.
1131. 1131 .merge(
1132. 1132 axum::Router::new()
1133. 1133 .merge(admin_ui::routes())
1134. 1134 .merge(client_ui::routes())
1135. 1135 .merge(site::routes())
1136. 1136 .layer(tower_http::compression::CompressionLayer::new().gzip(true)),
1137. 1137 )
1138. 1138 .with_state(state)
1139. 1139}
1140. 1140
1141. 1141/// Service-control plane over vak-ops: lets TUI/desktop/tray agree on the
1142. 1142/// same truth (docs/design/28-operations.md).
1143. 1143fn ops_payload(cfg: &vak_ops::OpsConfig) -> serde_json::Value {
1144. 1144 let st = |svc| vak_ops::status(svc, cfg);
1145. 1145 serde_json::json!({
1146. 1146 "gateway": { "state": st(vak_ops::Service::Gateway).to_string() },
1147. 1147 "bridges": { "state": st(vak_ops::Service::Bridges).to_string() },
1148. 1148 "gateway_healthy": vak_ops::health_ok(cfg),
1149. 1149 })
1150. 1150}
1151. 1151
1152. 1152async fn ops_status(State(state): State<AppState>) -> Json<serde_json::Value> {
1153. 1153 let mut cfg = vak_ops::OpsConfig::detect();
1154. 1154 cfg.port = state.ops_port;
1155. 1155 let payload = tokio::task::spawn_blocking(move || ops_payload(&cfg))
1156. 1156 .await
1157. 1157 .unwrap_or_else(|_| {
1158. 1158 serde_json::json!({
1159. 1159 "gateway": { "state": "unknown" },
1160. 1160 "telegram": { "state": "unknown" },
1161. 1161 "gateway_healthy": false,
1162. 1162 })
1163. 1163 });
1164. 1164 Json(payload)
1165. 1165}
1166. 1166
1167. 1167/// Read-only operational projection for desktop/TUI surfaces. This keeps
1168. 1168/// service, gateway, flow and health state in one refreshable payload without
1169. 1169/// exposing credentials or implementation paths.
1170. 1170async fn ops_diagnostics(State(state): State<AppState>) -> Json<serde_json::Value> {
1171. 1171 refresh_control_plane(&state);
1172. 1172 let mut cfg = vak_ops::OpsConfig::detect();
1173. 1173 cfg.port = state.ops_port;
1174. 1174 let root = state.core.sessions_home().join("flow-runs");
1175. 1175 let mut flows = Vec::new();
1176. 1176 if let Ok(entries) = std::fs::read_dir(&root) {
1177. 1177 for entry in entries.flatten().filter(|e| e.path().is_dir()) {
1178. 1178 let name = entry.file_name().to_string_lossy().into_owned();
1179. 1179 let runs = std::fs::read_dir(entry.path())
1180. 1180 .map(|items| {
1181. 1181 items
1182. 1182 .flatten()
1183. 1183 .filter(|e| e.path().extension().is_some_and(|x| x == "json"))
1184. 1184 .count()
1185. 1185 })
1186. 1186 .unwrap_or(0);
1187. 1187 flows.push(serde_json::json!({ "name": name, "runs": runs }));
1188. 1188 }
1189. 1189 }
1190. 1190 flows.sort_by(|a, b| a["name"].as_str().cmp(&b["name"].as_str()));
1191. 1191 let gateway = state.gateway.snapshot();
1192. 1192 let services = tokio::task::spawn_blocking(move || ops_payload(&cfg))
1193. 1193 .await
1194. 1194 .unwrap_or_else(|_| {
1195. 1195 serde_json::json!({
1196. 1196 "gateway": { "state": "unknown" },
1197. 1197 "telegram": { "state": "unknown" },
1198. 1198 "gateway_healthy": false,
1199. 1199 })
1200. 1200 });
1201. 1201 Json(serde_json::json!({
1202. 1202 "health": health_projection(&state),
1203. 1203 "services": services,
1204. 1204 "gateway": {
1205. 1205 "enabled": state.gateway.enabled,
1206. 1206 "bindings": gateway.into_iter().map(|(target, binding)| serde_json::json!({
1207. 1207 "target": target,
1208. 1208 "session_id": binding.session_id,
1209. 1209 "provider": binding.provider,
1210. 1210 "model": binding.model,
1211. 1211 "workspace": binding.workspace,
1212. 1212 "route_revision": binding.route_revision,
1213. 1213 })).collect::<Vec<_>>(),
1214. 1214 "approvals": {
1215. 1215 "mode": state.gateway.approvals_mode(),
1216. 1216 "approver": state.gateway.approver_target(),
1217. 1217 "pending": state.gateway.pending_approval_count(),
1218. 1218 },
1219. 1219 },
1220. 1220 "flows": flows,
1221. 1221 }))
1222. 1222}
1223. 1223
1224. 1224fn operation_services(cfg: &vak_ops::OpsConfig) -> serde_json::Value {
1225. 1225 let gateway = vak_ops::status(vak_ops::Service::Gateway, cfg);
1226. 1226 let bridges = vak_ops::status(vak_ops::Service::Bridges, cfg);
1227. 1227 serde_json::json!({
1228. 1228 "gateway": { "state": gateway.to_string() },
1229. 1229 "bridges": { "state": bridges.to_string() },
1230. 1230 "gateway_healthy": vak_ops::health_ok(cfg),
1231. 1231 })
1232. 1232}
1233. 1233
1234. 1234fn operation_runs(state: &AppState) -> Vec<serde_json::Value> {
1235. 1235 state
1236. 1236 .live_handles()
1237. 1237 .into_iter()
1238. 1238 .filter_map(|handle| {
1239. 1239 let active = handle
1240. 1240 .session
1241. 1241 .lock()
1242. 1242 .unwrap_or_else(std::sync::PoisonError::into_inner)
1243. 1243 .is_none();
1244. 1244 let pending = handle
1245. 1245 .pending
1246. 1246 .lock()
1247. 1247 .unwrap_or_else(std::sync::PoisonError::into_inner)
1248. 1248 .values()
1249. 1249 .map(|request| {
1250. 1250 serde_json::json!({
1251. 1251 "id": request.id,
1252. 1252 "tool": request.tool,
1253. 1253 "reason": request.reason,
1254. 1254 "requested_at": request.requested_at,
1255. 1255 })
1256. 1256 })
1257. 1257 .collect::<Vec<_>>();
1258. 1258 if !active && pending.is_empty() {
1259. 1259 return None;
1260. 1260 }
1261. 1261 let (agent_id, agent_name) = handle
1262. 1262 .core
1263. 1263 .agent_identity()
1264. 1264 .map(|id| (id.id.clone(), id.name.clone()))
1265. 1265 .unwrap_or_else(|| ("vak".to_string(), "Vakyartha".to_string()));
1266. 1266 Some(serde_json::json!({
1267. 1267 "session_id": handle.id,
1268. 1268 "workspace": handle.cwd,
1269. 1269 "agent_id": agent_id,
1270. 1270 "agent_name": agent_name,
1271. 1271 "state": if !pending.is_empty() { "waiting_approval" } else { "running" },
1272. 1272 "pending_approvals": pending,
1273. 1273 }))
1274. 1274 })
1275. 1275 .collect()
1276. 1276}
1277. 1277
1278. 1278fn operation_tasks(state: &AppState) -> Vec<serde_json::Value> {
1279. 1279 let tasks = state
1280. 1280 .tasks
1281. 1281 .lock()
1282. 1282 .unwrap_or_else(std::sync::PoisonError::into_inner);
1283. 1283 let next_fire = state
1284. 1284 .next_fire
1285. 1285 .lock()
1286. 1286 .unwrap_or_else(std::sync::PoisonError::into_inner);
1287. 1287 let inflight = state
1288. 1288 .script_inflight
1289. 1289 .lock()
1290. 1290 .unwrap_or_else(std::sync::PoisonError::into_inner);
1291. 1291 tasks
1292. 1292 .values()
1293. 1293 .map(|task| {
1294. 1294 let mut value = serde_json::to_value(task).unwrap_or_else(|_| serde_json::json!({}));
1295. 1295 if let Some(object) = value.as_object_mut() {
1296. 1296 object.insert(
1297. 1297 "next_fire".into(),
1298. 1298 next_fire
1299. 1299 .get(&task.id)
1300. 1300 .map(|fire| serde_json::Value::String(fire.to_rfc3339()))
1301. 1301 .unwrap_or(serde_json::Value::Null),
1302. 1302 );
1303. 1303 object.insert(
1304. 1304 "running".into(),
1305. 1305 serde_json::Value::Bool(inflight.contains(&task.id)),
1306. 1306 );
1307. 1307 }
1308. 1308 value
1309. 1309 })
1310. 1310 .collect()
1311. 1311}
1312. 1312
1313. 1313fn operation_outbox(state: &AppState) -> Result<(Vec<serde_json::Value>, usize, usize), String> {
1314. 1314 let records = delivery::outbox_records(&state.core)?;
1315. 1315 let pending = records
1316. 1316 .iter()
1317. 1317 .filter(|record| record.state == vak_delivery::outbox::OutboxState::Pending)
1318. 1318 .count();
1319. 1319 let dead = records
1320. 1320 .iter()
1321. 1321 .filter(|record| record.state == vak_delivery::outbox::OutboxState::DeadLetter)
1322. 1322 .count();
1323. 1323 let rows = records
1324. 1324 .into_iter()
1325. 1325 .take(200)
1326. 1326 .map(|record| {
1327. 1327 serde_json::json!({
1328. 1328 "job_id": record.job.job_id,
1329. 1329 "target": record.job.target,
1330. 1330 "kind": record.job.kind,
1331. 1331 "state": record.state,
1332. 1332 "attempts": record.attempts,
1333. 1333 "created_at_ms": record.created_at_ms,
1334. 1334 "updated_at_ms": record.updated_at_ms,
1335. 1335 "last_error": record.last_error,
1336. 1336 })
1337. 1337 })
1338. 1338 .collect();
1339. 1339 Ok((rows, pending, dead))
1340. 1340}
1341. 1341
1342. 1342/// Unified, evidence-backed projection for the Operations Center. Every row
1343. 1343/// is derived from an existing ledger, manager probe, or in-process handle;
1344. 1344/// unavailable state stays explicit instead of being painted green.
1345. 1345async fn operations_center(State(state): State<AppState>) -> Json<serde_json::Value> {
1346. 1346 refresh_control_plane(&state);
1347. 1347 load_tasks(&state);