Vakyartha documentation

327 return unwrap_mcp_response(response);
328 }
329 }
330
331 Err((
332 StatusCode::INTERNAL_SERVER_ERROR,
333 "No valid response from MCP server".into(),
334 ))
335}
336
337/// Unwrap a JSON-RPC `tools/call` response into the plain structured payload
338/// the frontend expects (`{items: [...]}`, `{alerts: [...]}`, etc.) instead of
339/// the raw MCP envelope (`{jsonrpc, id, result: {content, structuredContent}}`).
340fn unwrap_mcp_response(response: Value) -> Result<Value, (StatusCode, String)> {
341 if let Some(err) = response.get("error") {
342 let msg = err
343 .get("message")
344 .and_then(|m| m.as_str())
345 .unwrap_or("MCP error");
346 return Err((StatusCode::INTERNAL_SERVER_ERROR, msg.to_string()));
347 }
348
349 let result = response.get("result").cloned().unwrap_or(json!({}));
350
351 if result
352 .get("isError")
353 .and_then(|v| v.as_bool())
354 .unwrap_or(false)
355 {
356 let msg = result
357 .get("content")
358 .and_then(|c| c.get(0))
359 .and_then(|c| c.get("text"))
360 .and_then(|t| t.as_str())
361 .unwrap_or("MCP tool error");
362 return Err((StatusCode::INTERNAL_SERVER_ERROR, msg.to_string()));
363 }
364
365 if let Some(structured) = result.get("structuredContent") {
366 return Ok(structured.clone());
367 }
368
369 Ok(result)
370}
371
372/// GET /feeds/sources — List available source types from registry.
373pub async fn list_source_types() -> Result<impl IntoResponse, (StatusCode, String)> {
374 Ok(Json(json!({
375 "source_types": [
376 {"id": "rss", "name": "RSS / Atom Feed", "icon": "rss", "description": "Any RSS or Atom feed URL", "fetcher": "rss", "default_interval": "1h"},
377 {"id": "youtube", "name": "YouTube Channel", "icon": "youtube", "description": "Follow a channel's uploads", "fetcher": "youtube", "default_interval": "6h"},
378 {"id": "hacker_news", "name": "Hacker News", "icon": "fire", "description": "Top stories, best new, or Ask HN", "fetcher": "aggregator", "default_interval": "15m"},
379 {"id": "reddit", "name": "Reddit", "icon": "reddit", "description": "Follow subreddits", "fetcher": "aggregator", "default_interval": "30m"},
380 {"id": "lobsters", "name": "Lobste.rs", "icon": "lobsters", "description": "Lobste.rs technology stories", "fetcher": "aggregator", "default_interval": "30m"},
381 {"id": "custom_http", "name": "Custom HTTP Source", "icon": "globe", "description": "Any HTTP endpoint", "fetcher": "custom", "default_interval": "1h"},
382 ]
383 })))
384}
385
386/// GET /feeds/config — Get the effective scoped feed projection.
387pub async fn get_feed_config(
388 State(state): State<AppState>,
389) -> Result<impl IntoResponse, (StatusCode, String)> {
390 let cwd = state.core.cwd();
391 let sources = run_feed_mcp_request(
392 cwd,
393 &json!({"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"feed_sources","arguments":{}}}),
394 ).await?;
395 let alerts = run_feed_mcp_request(
396 cwd,
397 &json!({"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"feed_alerts","arguments":{}}}),
398 ).await?;
399 let stats = run_feed_script(cwd, "feed_ingest.py", &["--stats"]).await?;
400 Ok(Json(
401 json!({"sources": sources, "alerts": alerts, "stats": stats}),
402 ))
403}
404
405/// GET /feeds/items — List ingested items via MCP.
406pub async fn list_feed_items(
407 State(state): State<AppState>,
408 Query(params): Query<HashMap<String, String>>,
409) -> Result<impl IntoResponse, (StatusCode, String)> {
410 let cwd = state.core.cwd();
411 let limit = params
412 .get("limit")
413 .and_then(|s| s.parse::<usize>().ok())
414 .unwrap_or(50);
415 let source = params.get("source").map(|s| s.as_str()).unwrap_or("");
416
417 // Use the MCP feed_latest tool to get items
418 let mut arguments = json!({"limit": limit});
419 if !source.is_empty() {
420 arguments["source"] = json!(source);
421 }
422
423 let request = json!({
424 "jsonrpc": "2.0",
425 "id": 1,
426 "method": "tools/call",
427 "params": {
428 "name": "feed_latest",
429 "arguments": arguments
430 }
431 });
432
433 let result = run_feed_mcp_request(cwd, &request).await?;
434 let result = match result {
435 Value::Object(mut payload) => {
436 if let Some(items) = payload.remove("results") {
437 payload.insert("items".into(), items);
438 }
439 Value::Object(payload)
440 }
441 value => value,
442 };
443 Ok(Json(result))
444}
445
446/// GET /feeds/items/:id — Get single item.
447pub async fn get_feed_item(
448 State(state): State<AppState>,
449 Path(id): Path<i64>,
450) -> Result<impl IntoResponse, (StatusCode, String)> {
451 let cwd = state.core.cwd();
452 let request = json!({
453 "jsonrpc": "2.0",
454 "id": 1,
455 "method": "tools/call",
456 "params": {
457 "name": "feed_item",
458 "arguments": {"id": id}
459 }
460 });
461 let result = run_feed_mcp_request(cwd, &request).await?;
462 Ok(Json(result))
463}
464
465/// GET /feeds/search — Search feed items.
466pub async fn search_feed_items(
467 State(state): State<AppState>,
468 Query(params): Query<HashMap<String, String>>,
469) -> Result<impl IntoResponse, (StatusCode, String)> {
470 let query = params.get("q").map(|s| s.as_str()).unwrap_or("");
471 if query.is_empty() {
472 return Err((
473 StatusCode::BAD_REQUEST,
474 "Missing query parameter 'q'".into(),
475 ));
476 }
477
478 let cwd = state.core.cwd();
479 let mut arguments = json!({"query": query});
480
481 if let Some(tags) = params.get("tags") {
482 let tag_list: Vec<&str> = tags.split(',').map(|s| s.trim()).collect();
483 arguments["tags"] = json!(tag_list);
484 }
485 if let Some(since) = params.get("since") {
486 arguments["since"] = json!(since);
487 }
488 if let Some(limit) = params.get("limit") {
489 arguments["limit"] = json!(limit.parse::<usize>().unwrap_or(10));
490 }
491 if let Some(source) = params.get("source") {
492 arguments["source"] = json!(source);
493 }
494
495 let request = json!({
496 "jsonrpc": "2.0",
497 "id": 1,
498 "method": "tools/call",
499 "params": {
500 "name": "feed_search",
501 "arguments": arguments
502 }
503 });
504
505 let result = run_feed_mcp_request(cwd, &request).await?;
506 Ok(Json(result))
507}
508
509/// GET /feeds/stats — Get ingestion statistics.
510pub async fn get_feed_stats(
511 State(state): State<AppState>,
512) -> Result<impl IntoResponse, (StatusCode, String)> {
513 let cwd = state.core.cwd();
514 let result = run_feed_script(cwd, "feed_ingest.py", &["--stats"]).await?;
515 Ok(Json(result))
516}
517
518/// GET /feeds/alerts — List alert rules via MCP.
519pub async fn get_feed_alerts(
520 State(state): State<AppState>,
521) -> Result<impl IntoResponse, (StatusCode, String)> {
522 let cwd = state.core.cwd();
523 run_feed_script(cwd, "feed_ingest.py", &["--sync-alerts"]).await?;
524 let request = json!({
525 "jsonrpc": "2.0",
526 "id": 1,
527 "method": "tools/call",
528 "params": {
529 "name": "feed_alerts",
530 "arguments": {}
531 }
532 });
533 let result = run_feed_mcp_request(cwd, &request).await?;
534 Ok(Json(result))
535}
536
537/// GET /feeds/runs — List recent ingestion receipts for the active workspace.
538pub async fn get_feed_runs(
539 State(state): State<AppState>,
540) -> Result<impl IntoResponse, (StatusCode, String)> {
541 let cwd = state.core.cwd();
542 let request = json!({
543 "jsonrpc": "2.0",
544 "id": 1,
545 "method": "tools/call",
546 "params": {"name": "feed_runs", "arguments": {"limit": 20}}
547 });
548 Ok(Json(run_feed_mcp_request(cwd, &request).await?))
549}
550
551pub async fn get_feed_quarantine(
552 State(state): State<AppState>,
553) -> Result<impl IntoResponse, (StatusCode, String)> {
554 let cwd = state.core.cwd();
555 let request = json!({
556 "jsonrpc": "2.0", "id": 1, "method": "tools/call",
557 "params": {"name": "feed_quarantine", "arguments": {"limit": 50}}
558 });
559 Ok(Json(run_feed_mcp_request(cwd, &request).await?))
560}
561
562pub async fn release_feed_item(
563 Path(id): Path<i64>,
564 State(state): State<AppState>,
565) -> Result<impl IntoResponse, (StatusCode, String)> {
566 authorize_feed_mutation(&state, "workspace")?;
567 let cwd = state.core.cwd();
568 let id_text = id.to_string();
569 let result =
570 run_feed_admin_script(cwd, "feed_ingest.py", &["--release-item", &id_text]).await?;
571 if result.get("status").and_then(Value::as_str) != Some("ok") {
572 return Err((StatusCode::NOT_FOUND, format!("Item {id} not found")));
573 }
574 Ok(Json(result))
575}
576
577/// POST /feeds/ingest — Trigger manual ingestion.
578pub async fn trigger_ingestion(
579 State(state): State<AppState>,
580) -> Result<impl IntoResponse, (StatusCode, String)> {
581 authorize_feed_mutation(&state, "workspace")?;
582 let cwd = state.core.cwd();
583 let workspace = cwd.to_string_lossy();
584 let mut result = run_feed_script(cwd, "feed_ingest.py", &["--workspace", &workspace]).await?;
585 let intents = result
586 .get("delivery_intents")
587 .and_then(Value::as_array)
588 .cloned()
589 .unwrap_or_default();
590 if !intents.is_empty() {
591 let delivered = deliver_and_ack_alerts(&state, &intents)
592 .await
593 .map_err(|error| (StatusCode::BAD_GATEWAY, error))?;
594 result["alerts_delivered"] = json!(delivered);
595 }
596 Ok(Json(result))
597}
598
599pub async fn scheduled_ingestion(state: &AppState) {
600 if !state.core.config().feeds.enabled {
601 return;
602 }
603 if matches!(
604 state.core.effective_permission_mode(),
605 vak_config::PermissionMode::ReadOnly
606 ) {
607 return;
608 }
609 let cwd = state.core.cwd();
610 let workspace = cwd.to_string_lossy();
611 let Ok(mut result) = run_feed_script(cwd, "feed_ingest.py", &["--workspace", &workspace]).await
612 else {
613 return;
614 };
615 let intents = result
616 .get("delivery_intents")
617 .and_then(Value::as_array)
618 .cloned()
619 .unwrap_or_default();
620 if intents.is_empty() {
621 return;
622 }
623 if let Ok(delivered) = deliver_and_ack_alerts(state, &intents).await {
624 result["alerts_delivered"] = json!(delivered);
625 }
626}
627
628async fn deliver_and_ack_alerts(state: &AppState, intents: &[Value]) -> Result<usize, String> {
629 let delivered = crate::delivery::deliver_feed_intents(&state.core, intents).await?;
630 for intent in intents {
631 let Some(alert_id) = intent.get("alert_id").and_then(Value::as_i64) else {
632 continue;
633 };
634 let Some(item_id) = intent
635 .get("item")
636 .and_then(|item| item.get("id"))
637 .and_then(Value::as_i64)
638 else {
639 continue;
640 };
641 let alert_text = alert_id.to_string();
642 let item_text = item_id.to_string();
643 let result = run_feed_admin_script(
644 state.core.cwd(),
645 "feed_ingest.py",
646 &["--mark-alert-delivered", &alert_text, &item_text],
647 )
648 .await
649 .map_err(|(_, error)| error)?;
650 if result.get("status").and_then(Value::as_str) != Some("ok") {
651 return Err(format!(
652 "could not acknowledge alert {alert_id} for item {item_id}"
653 ));
654 }
655 }
656 Ok(delivered)
657}
658
659/// POST /feeds/sources — Add a new source to feeds.toml.
660pub async fn add_feed_source(
661 State(state): State<AppState>,
662 Json(payload): Json<Value>,
663) -> Result<impl IntoResponse, (StatusCode, String)> {
664 let cwd = state.core.cwd();
665 let scope = payload
666 .get("scope")
667 .and_then(Value::as_str)
668 .unwrap_or("workspace");
669 authorize_feed_mutation(&state, scope)?;
670 let config_path = if scope == "global" {
671 global_feeds_config_path()
672 } else {
673 feeds_config_path(cwd)
674 };
675
676 // Extract fields from payload
677 let name = payload
678 .get("name")
679 .and_then(|v| v.as_str())
680 .unwrap_or("Untitled");
681 let source_type = payload
682 .get("type")
683 .and_then(|v| v.as_str())
684 .unwrap_or("rss");
685 let url = payload.get("url").and_then(|v| v.as_str()).unwrap_or("");
686 if !url.is_empty() {
687 validate_source_url(url).map_err(|error| (StatusCode::BAD_REQUEST, error))?;
688 }
689 let interval = payload
690 .get("interval")
691 .and_then(|v| v.as_str())
692 .unwrap_or("1h");
693 let tags = payload.get("tags").and_then(|v| v.as_array()).map(|a| {
694 a.iter()
695 .filter_map(|v| v.as_str())
696 .collect::<Vec<_>>()
697 .join(", ")
698 });
699 let trust = payload
700 .get("trust")
701 .and_then(|v| v.as_str())
702 .unwrap_or("medium");
703
704 let source_id = payload
705 .get("id")
706 .and_then(Value::as_str)
707 .filter(|id| !id.trim().is_empty())
708 .map(ToOwned::to_owned)
709 .unwrap_or_else(|| format!("src-{}", uuid::Uuid::now_v7().simple()));
710
711 // Build the new source block
712 let mut source_block = format!(
713 "\n[[sources]]\nid = \"{}\"\nname = \"{}\"\ntype = \"{}\"\n",
714 escape_toml(&source_id),
715 escape_toml(name),
716 escape_toml(source_type)
717 );
718 if !url.is_empty() {
719 source_block.push_str(&format!("url = \"{}\"\n", escape_toml(url)));
720 }
721 source_block.push_str(&format!("interval = \"{}\"\n", escape_toml(interval)));
722 if let Some(tags_str) = &tags
723 && !tags_str.is_empty()
724 {
725 source_block.push_str(&format!(
726 "tags = [{}]\n",
727 tags_str
728 .split(", ")
729 .map(|t| format!("\"{}\"", escape_toml(t)))
730 .collect::<Vec<_>>()
731 .join(", ")
732 ));
733 }
734 source_block.push_str(&format!("trust = \"{}\"\n", escape_toml(trust)));
735 source_block.push_str("enabled = true\n");
736
737 let source_id_line = format!("id = \"{}\"", escape_toml(&source_id));
738 let conflict = format!("source id '{source_id}' already exists in {scope} scope");
739 edit_feeds_config(config_path.clone(), move |current| {
740 let mut content = current.map_or_else(default_feeds_config, ToOwned::to_owned);
741 if content.lines().any(|line| line.trim() == source_id_line) {
742 return Err((StatusCode::CONFLICT, conflict));
743 }
744 append_block(&mut content, &source_block);
745 Ok((Some(content), ()))
746 })
747 .await?;
748
749 Ok(Json(json!({
750 "status": "ok",
751 "message": format!("Source '{}' added to {}", name, config_path.display()),
752 "source": payload
753 })))
754}
755
756/// Escape a string for TOML output.
757fn escape_toml(s: &str) -> String {
758 s.replace('\\', "\\\\")
759 .replace('"', "\\\"")
760 .replace('\n', "\\n")
761}
762
763/// Find the line range `[start, end)` of a `[[table]]` array-of-tables block
764/// whose `name = "..."` field matches `name`. `end` is the index of the first
765/// line after the block: the next `[[...]]` header, a `[section]` header that
766/// isn't a dotted sub-table of this array item (e.g. `[alerts.match]` stays
767/// part of the `[[alerts]]` block it follows), or EOF.
768fn find_array_table_block(lines: &[&str], table: &str, name: &str) -> Option<(usize, usize)> {
769 let header = format!("[[{}]]", table);
770 let subtable_prefix = format!("[{}.", table);
771 let mut i = 0;
772 while i < lines.len() {
773 if lines[i].trim() == header {
774 let start = i;
775 let mut j = i + 1;
776 let mut block_key: Option<String> = None;
777 while j < lines.len() {
778 let trimmed = lines[j].trim();
779 let is_boundary = trimmed.starts_with("[[")
780 || (trimmed.starts_with('[') && !trimmed.starts_with(&subtable_prefix));
781 if is_boundary {
782 break;
783 }
784 if let Some(rest) = trimmed.strip_prefix("id") {
785 let rest = rest.trim_start();
786 if let Some(rest) = rest.strip_prefix('=') {
787 block_key = Some(unescape_toml(rest.trim().trim_matches('"')));
788 }
789 } else if block_key.is_none()
790 && let Some(rest) = trimmed.strip_prefix("name")
791 {
792 let rest = rest.trim_start();
793 if let Some(rest) = rest.strip_prefix('=') {
794 let value = rest.trim().trim_matches('"');
795 block_key = Some(unescape_toml(value));
796 }
797 }
798 j += 1;
799 }
800 if block_key.as_deref() == Some(name) {
801 return Some((start, j));
802 }
803 i = j;
804 } else {
805 i += 1;
806 }
807 }
808 None
809}
810
811/// Reverse of `escape_toml` for simple double-quoted TOML string values.
812fn unescape_toml(s: &str) -> String {
813 s.replace("\\\"", "\"").replace("\\\\", "\\")
814}
815
816/// Remove a `[[table]]` block by name from TOML `content`. Returns the new
817/// content, or `None` if no matching block was found.
818fn remove_array_table_block(content: &str, table: &str, name: &str) -> Option<String> {
819 let lines: Vec<&str> = content.lines().collect();
820 let (start, end) = find_array_table_block(&lines, table, name)?;
821 let mut new_lines: Vec<&str> = Vec::with_capacity(lines.len());
822 new_lines.extend_from_slice(&lines[..start]);
823 new_lines.extend_from_slice(&lines[end..]);
824 Some(new_lines.join("\n"))
825}
826
827/// Set (or add) a scalar `field = value` line inside a `[[table]]` block
828/// matched by name, replacing the existing line for that field if present.
829fn set_field_in_block(
830 content: &str,
831 table: &str,
832 name: &str,
833 field: &str,
834 value_line: &str,
835) -> Option<String> {
836 let lines: Vec<&str> = content.lines().collect();
837 let (start, end) = find_array_table_block(&lines, table, name)?;
838 let mut new_lines: Vec<String> = lines[..=start].iter().map(|l| l.to_string()).collect();
839 let mut replaced = false;
840 for line in &lines[start + 1..end] {
841 let trimmed = line.trim();
842 if trimmed.starts_with(field) && trimmed[field.len()..].trim_start().starts_with('=') {
843 new_lines.push(value_line.to_string());
844 replaced = true;
845 } else {
846 new_lines.push(line.to_string());
847 }
848 }
849 if !replaced {
850 new_lines.push(value_line.to_string());
851 }
852 new_lines.extend(lines[end..].iter().map(|l| l.to_string()));
853 Some(new_lines.join("\n"))
854}
855
856fn default_feeds_config() -> String {
857 "[general]\ndefault_check_interval = \"30m\"\nmax_items_per_feed = 500\ndedup_window_days = 90\n\n".to_string()
858}
859
860fn append_block(content: &mut String, block: &str) {
861 if !content.is_empty() && !content.ends_with('\n') {
862 content.push('\n');
863 }
864 content.push_str(block);
865}
866
867/// Every edit to a `feeds.toml` goes through here: `edit` sees the current
868/// text (`None` when the file does not exist) and returns the new text, or
869/// `None` to leave it alone. The read, the edit and the atomic replace hold
870/// one lock (`vak_config::file_update`), so concurrent edits never publish a
871/// half-written file or rewrite the file from a stale read.
872async fn edit_feeds_config<T: Send + 'static>(
873 config_path: PathBuf,
874 edit: impl FnOnce(Option<&str>) -> Result<(Option<String>, T), (StatusCode, String)>
875 + Send
876 + 'static,
877) -> Result<T, (StatusCode, String)> {
878 tokio::task::spawn_blocking(move || {
879 vak_config::file_update::update_file(&config_path, edit).map_err(|error| match error {
880 vak_config::file_update::UpdateError::Edit(error) => error,
881 vak_config::file_update::UpdateError::Io { path, source } => (
882 StatusCode::INTERNAL_SERVER_ERROR,
883 format!("Failed to update {}: {source}", path.display()),
884 ),
885 })
886 })
887 .await
888 .map_err(|error| {
889 (
890 StatusCode::INTERNAL_SERVER_ERROR,
891 format!("Feed config update did not finish: {error}"),
892 )
893 })?
894}
895
896/// DELETE /feeds/sources/{name} — Remove a feed source from config.
897/// DELETE /feeds/sources/{name} — Remove a feed source.
898///
899/// Soft-deletes the materialized source row and removes its declaration from
900/// the canonical scope-specific configuration.
901pub async fn delete_feed_source(
902 Path(source_id): Path<String>,
903 Query(params): Query<HashMap<String, String>>,
904 State(state): State<AppState>,
905) -> Result<impl IntoResponse, (StatusCode, String)> {
906 let cwd = state.core.cwd();
907 let scope = params
908 .get("scope")
909 .map(String::as_str)
910 .unwrap_or("workspace");
911 authorize_feed_mutation(&state, scope)?;
912
913 let removed = run_feed_admin_script(
914 cwd,
915 "feed_ingest.py",
916 &["--remove-source", &source_id, "--scope", scope],
917 )
918 .await?
919 .get("status")
920 .and_then(|s| s.as_str())
921 == Some("ok");
922
923 if !removed {
924 return Err((
925 StatusCode::NOT_FOUND,
926 format!("Source '{}' not found", source_id),
927 ));
928 }
929
930 let config_path = if scope == "global" {
931 global_feeds_config_path()
932 } else {
933 feeds_config_path(cwd)
934 };
935 let declared = source_id.clone();
936 let _ = edit_feeds_config(config_path, move |current| {
937 let next =
938 current.and_then(|content| remove_array_table_block(content, "sources", &declared));
939 Ok((next, ()))
940 })
941 .await;
942
943 Ok(Json(json!({
944 "status": "ok",
945 "message": format!("Source '{}' removed", source_id),
946 })))
947}
948
949/// PATCH /feeds/sources/{name} — Update a source's enabled/interval/trust fields.
950///
951/// `tags` is **refused** here. The `feeds` table has no tags column, so a
952/// tag written on update reached feeds.toml and nothing else: the DB never
953/// saw it and the UI never showed it. A PATCH that silently half-applies
954/// is worse than one that says no, so the caller is told where tags are
955/// actually set instead (AGENTS.md invariant 30).
956pub async fn update_feed_source(
957 Path(source_id): Path<String>,
958 Query(params): Query<HashMap<String, String>>,
959 State(state): State<AppState>,
960 Json(payload): Json<Value>,
961) -> Result<impl IntoResponse, (StatusCode, String)> {
962 let cwd = state.core.cwd();
963 let scope = params
964 .get("scope")
965 .map(String::as_str)
966 .unwrap_or("workspace");
967 authorize_feed_mutation(&state, scope)?;
968
969 let enabled = payload.get("enabled").and_then(|v| v.as_bool());
970 let interval = payload.get("interval").and_then(|v| v.as_str());
971 let trust = payload.get("trust").and_then(|v| v.as_str());
972 if payload.get("tags").is_some() {
973 return Err((
974 StatusCode::BAD_REQUEST,
975 "tags cannot be updated here: the feeds table has no tags column, so the \
976 change would reach feeds.toml and nothing else. Set tags when creating \
977 the source."
978 .into(),
979 ));
980 }
981
982 if enabled.is_none() && interval.is_none() && trust.is_none() {
983 return Err((
984 StatusCode::BAD_REQUEST,
985 "No recognized fields to update (enabled, interval, trust)".into(),
986 ));
987 }
988
989 if enabled.is_some() || interval.is_some() || trust.is_some() {
990 let mut args: Vec<String> = vec!["--update-source".into(), source_id.clone()];
991 args.extend(["--scope".into(), scope.into()]);
992 if let Some(e) = enabled {
993 args.push("--set-enabled".into());
994 args.push(if e { "true" } else { "false" }.into());
995 }
996 if let Some(i) = interval {
997 args.push("--set-interval".into());
998 args.push(i.into());
999 }
1000 if let Some(t) = trust {
1001 args.push("--set-trust".into());
1002 args.push(t.into());
1003 }
1004 let arg_refs: Vec<&str> = args.iter().map(|s| s.as_str()).collect();
1005 let updated = run_feed_admin_script(cwd, "feed_ingest.py", &arg_refs)
1006 .await?
1007 .get("status")
1008 .and_then(|s| s.as_str())
1009 == Some("ok");
1010 if !updated {
1011 return Err((
1012 StatusCode::NOT_FOUND,
1013 format!("Source '{}' not found", source_id),
1014 ));
1015 }
1016 }
1017
1018 // Reconcile the durable declaration when this source has one. The
1019 // materialized row above remains authoritative for the current response.
1020 let config_path = if scope == "global" {
1021 global_feeds_config_path()
1022 } else {
1023 feeds_config_path(cwd)
1024 };
1025 let mut fields = Vec::new();
1026 if let Some(e) = enabled {
1027 fields.push(("enabled", format!("enabled = {e}")));
1028 }
1029 if let Some(i) = interval {
1030 fields.push(("interval", format!("interval = \"{}\"", escape_toml(i))));
1031 }
1032 if let Some(t) = trust {
1033 fields.push(("trust", format!("trust = \"{}\"", escape_toml(t))));
1034 }
1035 let declared = source_id.clone();
1036 let _ = edit_feeds_config(config_path, move |current| {
1037 let Some(mut content) = current.map(ToOwned::to_owned) else {
1038 return Ok((None, ()));
1039 };
1040 let mut touched = false;
1041 for (field, line) in &fields {
1042 if let Some(next) = set_field_in_block(&content, "sources", &declared, field, line) {
1043 content = next;
1044 touched = true;
1045 }
1046 }
1047 Ok((touched.then_some(content), ()))
1048 })
1049 .await;
1050
1051 Ok(Json(json!({
1052 "status": "ok",
1053 "message": format!("Source '{}' updated", source_id),
1054 })))
1055}
1056
1057/// GET /feeds/sources/configured — List sources actually known to the DB
1058/// (populated by ingestion), with live enabled/trust/interval status.
1059pub async fn list_configured_sources(
1060 State(state): State<AppState>,
1061) -> Result<impl IntoResponse, (StatusCode, String)> {
1062 let cwd = state.core.cwd();
1063 let request = json!({
1064 "jsonrpc": "2.0",
1065 "id": 1,
1066 "method": "tools/call",
1067 "params": {
1068 "name": "feed_sources",
1069 "arguments": {}
1070 }
1071 });
1072 let result = run_feed_mcp_request(cwd, &request).await?;
1073 Ok(Json(result))
1074}
1075
1076/// POST /feeds/alerts — Add a new alert rule to feeds.toml.
1077pub async fn add_feed_alert(
1078 State(state): State<AppState>,
1079 Json(payload): Json<Value>,
1080) -> Result<impl IntoResponse, (StatusCode, String)> {
1081 let cwd = state.core.cwd();
1082 let scope = payload
1083 .get("scope")
1084 .and_then(Value::as_str)
1085 .unwrap_or("workspace");
1086 authorize_feed_mutation(&state, scope)?;
1087 let config_path = if scope == "global" {
1088 global_feeds_config_path()
1089 } else {
1090 feeds_config_path(cwd)
1091 };
1092
1093 let name = payload
1094 .get("name")
1095 .and_then(|v| v.as_str())
1096 .filter(|s| !s.is_empty())
1097 .ok_or((StatusCode::BAD_REQUEST, "Alert 'name' is required".into()))?;
1098
1099 let str_array = |key: &str| -> Vec<String> {
1100 payload
1101 .get(key)
1102 .and_then(|v| v.as_array())
1103 .map(|a| {
1104 a.iter()
1105 .filter_map(|v| v.as_str())
1106 .map(String::from)
1107 .collect()
1108 })
1109 .unwrap_or_default()
1110 };
1111 let keywords = str_array("keywords");
1112 let tags = str_array("tags");
1113 let sources = str_array("sources");
1114
1115 if keywords.is_empty() && tags.is_empty() && sources.is_empty() {
1116 return Err((
1117 StatusCode::BAD_REQUEST,
1118 "At least one of keywords, tags, or sources must be set".into(),
1119 ));
1120 }
1121
1122 let action = payload
1123 .get("action")
1124 .and_then(|v| v.as_str())
1125 .unwrap_or("deliver");
1126 if !matches!(action, "deliver" | "hook" | "both") {
1127 return Err((
1128 StatusCode::BAD_REQUEST,
1129 "action must be one of: deliver, hook, both".into(),
1130 ));
1131 }
1132 let deliver_to = payload
1133 .get("deliver_to")
1134 .and_then(|v| v.as_str())
1135 .unwrap_or("");
1136 if matches!(action, "deliver" | "both") && deliver_to.trim().is_empty() {
1137 return Err((
1138 StatusCode::BAD_REQUEST,
1139 "deliver_to is required for deliver and both alerts".into(),
1140 ));
1141 }
1142 let cooldown_minutes = payload
1143 .get("cooldown_minutes")
1144 .and_then(|v| v.as_i64())
1145 .unwrap_or(30);
1146
1147 let to_toml_array = |items: &[String]| -> String {
1148 items
1149 .iter()
1150 .map(|t| format!("\"{}\"", escape_toml(t)))
1151 .collect::<Vec<_>>()
1152 .join(", ")
1153 };
1154
1155 let mut block = format!(
1156 "\n[[alerts]]\nname = \"{}\"\naction = \"{}\"\ncooldown_minutes = {}\nenabled = true\n",
1157 escape_toml(name),
1158 escape_toml(action),
1159 cooldown_minutes
1160 );
1161 if !deliver_to.is_empty() {
1162 block.push_str(&format!("deliver_to = \"{}\"\n", escape_toml(deliver_to)));
1163 }
1164 block.push_str("\n[alerts.match]\n");
1165 if !keywords.is_empty() {
1166 block.push_str(&format!("keywords = [{}]\n", to_toml_array(&keywords)));
1167 }
1168 if !tags.is_empty() {
1169 block.push_str(&format!("tags = [{}]\n", to_toml_array(&tags)));
1170 }
1171 if !sources.is_empty() {
1172 block.push_str(&format!("sources = [{}]\n", to_toml_array(&sources)));
1173 }
1174
1175 edit_feeds_config(config_path.clone(), move |current| {
1176 let mut content = current.map_or_else(default_feeds_config, ToOwned::to_owned);
1177 append_block(&mut content, &block);
1178 Ok((Some(content), ()))
1179 })
1180 .await?;
1181 run_feed_script(cwd, "feed_ingest.py", &["--sync-alerts"]).await?;
1182
1183 Ok(Json(json!({
1184 "status": "ok",
1185 "message": format!("Alert '{}' added to {}", name, config_path.display()),
1186 })))
1187}
1188
1189/// DELETE /feeds/alerts/{name} — Remove an alert rule from config.
1190pub async fn delete_feed_alert(
1191 Path(name): Path<String>,
1192 Query(params): Query<HashMap<String, String>>,
1193 State(state): State<AppState>,
1194) -> Result<impl IntoResponse, (StatusCode, String)> {
1195 let cwd = state.core.cwd();
1196 let scope = params
1197 .get("scope")
1198 .map(String::as_str)
1199 .unwrap_or("workspace");
1200 authorize_feed_mutation(&state, scope)?;
1201 let config_path = if scope == "global" {
1202 global_feeds_config_path()
1203 } else {
1204 feeds_config_path(cwd)
1205 };
1206
1207 let alert = name.clone();
1208 edit_feeds_config(config_path.clone(), move |current| {
1209 let content =
1210 current.ok_or_else(|| (StatusCode::NOT_FOUND, "Config file not found".to_string()))?;
1211 let next = remove_array_table_block(content, "alerts", &alert)
1212 .ok_or_else(|| (StatusCode::NOT_FOUND, format!("Alert '{alert}' not found")))?;
1213 Ok((Some(next), ()))
1214 })
1215 .await?;
1216 run_feed_script(cwd, "feed_ingest.py", &["--sync-alerts"]).await?;
1217
1218 Ok(Json(json!({
1219 "status": "ok",
1220 "message": format!("Alert '{}' removed from {}", name, config_path.display()),
1221 })))
1222}
1223
1224/// Build feed routes.
1225pub fn routes() -> Router<AppState> {
1226 Router::new()
1227 .route(
1228 "/feeds/sources",
1229 get(list_source_types).post(add_feed_source),
1230 )
1231 .route("/feeds/sources/configured", get(list_configured_sources))
1232 .route(
1233 "/feeds/sources/{name}",
1234 delete(delete_feed_source).patch(update_feed_source),
1235 )
1236 .route("/feeds/config", get(get_feed_config))
1237 .route("/feeds/items", get(list_feed_items))
1238 .route("/feeds/items/{id}", get(get_feed_item))
1239 .route("/feeds/search", get(search_feed_items))
1240 .route("/feeds/stats", get(get_feed_stats))
1241 .route("/feeds/alerts", get(get_feed_alerts).post(add_feed_alert))
1242 .route("/feeds/runs", get(get_feed_runs))
1243 .route("/feeds/quarantine", get(get_feed_quarantine))
1244 .route("/feeds/quarantine/{id}/release", post(release_feed_item))
1245 .route("/feeds/alerts/{name}", delete(delete_feed_alert))
1246 .route("/feeds/ingest", post(trigger_ingestion))
1247}
1248
1249#[cfg(test)]
1250#[allow(clippy::panic, clippy::unwrap_used, clippy::expect_used)]
1251mod tests {
1252 use super::{authorize_feed_scope, feed_environment, feeds_dir, validate_source_url};
1253 use vak_config::PermissionMode;
1254
1255 /// The feed subprocess runs with `env_clear`, so `feed_environment` is the
1256 /// complete list of what it sees. It must carry the operational minimum
1257 /// (PATH, so `python3` resolves) and never a parent secret — before this,
1258 /// the subprocess inherited the whole server environment, `VAK_GATEWAY_TOKEN`
1259 /// and provider keys included (invariant 12).
1260 #[test]
1261 fn feed_subprocess_environment_is_a_secret_free_allowlist() {
1262 let _home = vak_config::paths::isolate_home_for_tests();
1263 let environment = feed_environment(
1264 std::path::Path::new("/tmp/ws"),
1265 std::path::Path::new("/tmp/scripts"),
1266 );
1267 let allowed = [
1268 "PYTHONPATH",
1269 "VAK_FEED_WORKSPACE",
1270 "VAK_FEEDS_DB",
1271 "VAK_FEEDS_LOG",
1272 "VAK_FEEDS_CONFIG",
1273 "PATH",
1274 "HOME",
1275 ];
1276 for (key, _) in &environment {
1277 assert!(allowed.contains(key), "{key} is not in the feed allowlist");
1278 }
1279 assert!(
1280 environment.iter().any(|(key, _)| *key == "PATH"),
1281 "PATH must be passed so python3 resolves under env_clear"
1282 );
1283 for secret in ["VAK_GATEWAY_TOKEN", "ANTHROPIC_API_KEY", "OPENAI_API_KEY"] {
1284 assert!(
1285 !environment.iter().any(|(key, _)| *key == secret),
1286 "{secret} must never be passed to a feed subprocess"
1287 );
1288 }
1289 }
1290
1291 /// The script the server runs is located only from the binary, never from
1292 /// the session workspace: a workspace with its own `scripts/feeds` cannot
1293 /// supply the code the server executes (invariants 12, 14, 15).
1294 #[test]
1295 fn feeds_dir_never_resolves_from_a_workspace() {
1296 let workspace = tempfile::tempdir().expect("tempdir");
1297 let planted = workspace.path().join("scripts").join("feeds");
1298 std::fs::create_dir_all(&planted).expect("mkdir");
1299 std::fs::write(planted.join("feed_ingest.py"), b"raise SystemExit\n").expect("write");
1300 assert!(
1301 !feeds_dir().starts_with(workspace.path()),
1302 "feeds_dir resolved a script from the session workspace"
1303 );
1304 }
1305
1306 /// Every path a feed script writes comes from the canonical data home,
1307 /// so an overridden `VAK_HOME` holds the feed store, its log and its
1308 /// config; the scripts used to work out the platform default for
1309 /// themselves and wrote outside it.
1310 #[test]
1311 fn feeds_write_under_overridden_home() {
1312 let home = vak_config::paths::isolate_home_for_tests();
1313 let environment = feed_environment(
1314 std::path::Path::new("/tmp/ws"),
1315 std::path::Path::new("/tmp/scripts"),
1316 );
1317 for name in ["VAK_FEEDS_DB", "VAK_FEEDS_LOG", "VAK_FEEDS_CONFIG"] {
1318 let (_, value) = environment
1319 .iter()
1320 .find(|(key, _)| *key == name)
1321 .unwrap_or_else(|| panic!("{name} is passed"));
1322 assert!(
1323 std::path::Path::new(value).starts_with(&home),
1324 "{name} = {value} is outside {}",
1325 home.display()
1326 );

327 return unwrap_mcp_response(response);
328 }
329 }
330
331 Err((
332 StatusCode::INTERNAL_SERVER_ERROR,
333 "No valid response from MCP server".into(),
334 ))
335}
336
337/// Unwrap a JSON-RPC `tools/call` response into the plain structured payload
338/// the frontend expects (`{items: [...]}`, `{alerts: [...]}`, etc.) instead of
339/// the raw MCP envelope (`{jsonrpc, id, result: {content, structuredContent}}`).
340fn unwrap_mcp_response(response: Value) -> Result<Value, (StatusCode, String)> {
341 if let Some(err) = response.get("error") {
342 let msg = err
343 .get("message")
344 .and_then(|m| m.as_str())
345 .unwrap_or("MCP error");
346 return Err((StatusCode::INTERNAL_SERVER_ERROR, msg.to_string()));
347 }
348
349 let result = response.get("result").cloned().unwrap_or(json!({}));
350
351 if result
352 .get("isError")
353 .and_then(|v| v.as_bool())
354 .unwrap_or(false)
355 {
356 let msg = result
357 .get("content")
358 .and_then(|c| c.get(0))
359 .and_then(|c| c.get("text"))
360 .and_then(|t| t.as_str())
361 .unwrap_or("MCP tool error");
362 return Err((StatusCode::INTERNAL_SERVER_ERROR, msg.to_string()));
363 }
364
365 if let Some(structured) = result.get("structuredContent") {
366 return Ok(structured.clone());
367 }
368
369 Ok(result)
370}
371
372/// GET /feeds/sources — List available source types from registry.
373pub async fn list_source_types() -> Result<impl IntoResponse, (StatusCode, String)> {
374 Ok(Json(json!({
375 "source_types": [
376 {"id": "rss", "name": "RSS / Atom Feed", "icon": "rss", "description": "Any RSS or Atom feed URL", "fetcher": "rss", "default_interval": "1h"},
377 {"id": "youtube", "name": "YouTube Channel", "icon": "youtube", "description": "Follow a channel's uploads", "fetcher": "youtube", "default_interval": "6h"},
378 {"id": "hacker_news", "name": "Hacker News", "icon": "fire", "description": "Top stories, best new, or Ask HN", "fetcher": "aggregator", "default_interval": "15m"},
379 {"id": "reddit", "name": "Reddit", "icon": "reddit", "description": "Follow subreddits", "fetcher": "aggregator", "default_interval": "30m"},
380 {"id": "lobsters", "name": "Lobste.rs", "icon": "lobsters", "description": "Lobste.rs technology stories", "fetcher": "aggregator", "default_interval": "30m"},
381 {"id": "custom_http", "name": "Custom HTTP Source", "icon": "globe", "description": "Any HTTP endpoint", "fetcher": "custom", "default_interval": "1h"},
382 ]
383 })))
384}
385
386/// GET /feeds/config — Get the effective scoped feed projection.
387pub async fn get_feed_config(
388 State(state): State<AppState>,
389) -> Result<impl IntoResponse, (StatusCode, String)> {
390 let cwd = state.core.cwd();
391 let sources = run_feed_mcp_request(
392 cwd,
393 &json!({"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"feed_sources","arguments":{}}}),
394 ).await?;
395 let alerts = run_feed_mcp_request(
396 cwd,
397 &json!({"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"feed_alerts","arguments":{}}}),
398 ).await?;
399 let stats = run_feed_script(cwd, "feed_ingest.py", &["--stats"]).await?;
400 Ok(Json(
401 json!({"sources": sources, "alerts": alerts, "stats": stats}),
402 ))
403}
404
405/// GET /feeds/items — List ingested items via MCP.
406pub async fn list_feed_items(
407 State(state): State<AppState>,
408 Query(params): Query<HashMap<String, String>>,
409) -> Result<impl IntoResponse, (StatusCode, String)> {
410 let cwd = state.core.cwd();
411 let limit = params
412 .get("limit")
413 .and_then(|s| s.parse::<usize>().ok())
414 .unwrap_or(50);
415 let source = params.get("source").map(|s| s.as_str()).unwrap_or("");
416
417 // Use the MCP feed_latest tool to get items
418 let mut arguments = json!({"limit": limit});
419 if !source.is_empty() {
420 arguments["source"] = json!(source);
421 }
422
423 let request = json!({
424 "jsonrpc": "2.0",
425 "id": 1,
426 "method": "tools/call",
427 "params": {
428 "name": "feed_latest",
429 "arguments": arguments
430 }
431 });
432
433 let result = run_feed_mcp_request(cwd, &request).await?;
434 let result = match result {
435 Value::Object(mut payload) => {
436 if let Some(items) = payload.remove("results") {
437 payload.insert("items".into(), items);
438 }
439 Value::Object(payload)
440 }
441 value => value,
442 };
443 Ok(Json(result))
444}
445
446/// GET /feeds/items/:id — Get single item.
447pub async fn get_feed_item(
448 State(state): State<AppState>,
449 Path(id): Path<i64>,
450) -> Result<impl IntoResponse, (StatusCode, String)> {
451 let cwd = state.core.cwd();
452 let request = json!({
453 "jsonrpc": "2.0",
454 "id": 1,
455 "method": "tools/call",
456 "params": {
457 "name": "feed_item",
458 "arguments": {"id": id}
459 }
460 });
461 let result = run_feed_mcp_request(cwd, &request).await?;
462 Ok(Json(result))
463}
464
465/// GET /feeds/search — Search feed items.
466pub async fn search_feed_items(
467 State(state): State<AppState>,
468 Query(params): Query<HashMap<String, String>>,
469) -> Result<impl IntoResponse, (StatusCode, String)> {
470 let query = params.get("q").map(|s| s.as_str()).unwrap_or("");
471 if query.is_empty() {
472 return Err((
473 StatusCode::BAD_REQUEST,
474 "Missing query parameter 'q'".into(),
475 ));
476 }
477
478 let cwd = state.core.cwd();
479 let mut arguments = json!({"query": query});
480
481 if let Some(tags) = params.get("tags") {
482 let tag_list: Vec<&str> = tags.split(',').map(|s| s.trim()).collect();
483 arguments["tags"] = json!(tag_list);
484 }
485 if let Some(since) = params.get("since") {
486 arguments["since"] = json!(since);
487 }
488 if let Some(limit) = params.get("limit") {
489 arguments["limit"] = json!(limit.parse::<usize>().unwrap_or(10));
490 }
491 if let Some(source) = params.get("source") {
492 arguments["source"] = json!(source);
493 }
494
495 let request = json!({
496 "jsonrpc": "2.0",
497 "id": 1,
498 "method": "tools/call",
499 "params": {
500 "name": "feed_search",
501 "arguments": arguments
502 }
503 });
504
505 let result = run_feed_mcp_request(cwd, &request).await?;
506 Ok(Json(result))
507}
508
509/// GET /feeds/stats — Get ingestion statistics.
510pub async fn get_feed_stats(
511 State(state): State<AppState>,
512) -> Result<impl IntoResponse, (StatusCode, String)> {
513 let cwd = state.core.cwd();
514 let result = run_feed_script(cwd, "feed_ingest.py", &["--stats"]).await?;
515 Ok(Json(result))
516}
517
518/// GET /feeds/alerts — List alert rules via MCP.
519pub async fn get_feed_alerts(
520 State(state): State<AppState>,
521) -> Result<impl IntoResponse, (StatusCode, String)> {
522 let cwd = state.core.cwd();
523 run_feed_script(cwd, "feed_ingest.py", &["--sync-alerts"]).await?;
524 let request = json!({
525 "jsonrpc": "2.0",
526 "id": 1,
527 "method": "tools/call",
528 "params": {
529 "name": "feed_alerts",
530 "arguments": {}
531 }
532 });
533 let result = run_feed_mcp_request(cwd, &request).await?;
534 Ok(Json(result))
535}
536
537/// GET /feeds/runs — List recent ingestion receipts for the active workspace.
538pub async fn get_feed_runs(
539 State(state): State<AppState>,
540) -> Result<impl IntoResponse, (StatusCode, String)> {
541 let cwd = state.core.cwd();
542 let request = json!({
543 "jsonrpc": "2.0",
544 "id": 1,
545 "method": "tools/call",
546 "params": {"name": "feed_runs", "arguments": {"limit": 20}}
547 });
548 Ok(Json(run_feed_mcp_request(cwd, &request).await?))
549}
550
551pub async fn get_feed_quarantine(
552 State(state): State<AppState>,
553) -> Result<impl IntoResponse, (StatusCode, String)> {
554 let cwd = state.core.cwd();
555 let request = json!({
556 "jsonrpc": "2.0", "id": 1, "method": "tools/call",
557 "params": {"name": "feed_quarantine", "arguments": {"limit": 50}}
558 });
559 Ok(Json(run_feed_mcp_request(cwd, &request).await?))
560}
561
562pub async fn release_feed_item(
563 Path(id): Path<i64>,
564 State(state): State<AppState>,
565) -> Result<impl IntoResponse, (StatusCode, String)> {
566 authorize_feed_mutation(&state, "workspace")?;
567 let cwd = state.core.cwd();
568 let id_text = id.to_string();
569 let result =
570 run_feed_admin_script(cwd, "feed_ingest.py", &["--release-item", &id_text]).await?;
571 if result.get("status").and_then(Value::as_str) != Some("ok") {
572 return Err((StatusCode::NOT_FOUND, format!("Item {id} not found")));
573 }
574 Ok(Json(result))
575}
576
577/// POST /feeds/ingest — Trigger manual ingestion.
578pub async fn trigger_ingestion(
579 State(state): State<AppState>,
580) -> Result<impl IntoResponse, (StatusCode, String)> {
581 authorize_feed_mutation(&state, "workspace")?;
582 let cwd = state.core.cwd();
583 let workspace = cwd.to_string_lossy();
584 let mut result = run_feed_script(cwd, "feed_ingest.py", &["--workspace", &workspace]).await?;
585 let intents = result
586 .get("delivery_intents")
587 .and_then(Value::as_array)
588 .cloned()
589 .unwrap_or_default();
590 if !intents.is_empty() {
591 let delivered = deliver_and_ack_alerts(&state, &intents)
592 .await
593 .map_err(|error| (StatusCode::BAD_GATEWAY, error))?;
594 result["alerts_delivered"] = json!(delivered);
595 }
596 Ok(Json(result))
597}
598
599pub async fn scheduled_ingestion(state: &AppState) {
600 if !state.core.config().feeds.enabled {
601 return;
602 }
603 if matches!(
604 state.core.effective_permission_mode(),
605 vak_config::PermissionMode::ReadOnly
606 ) {
607 return;
608 }
609 let cwd = state.core.cwd();
610 let workspace = cwd.to_string_lossy();
611 let Ok(mut result) = run_feed_script(cwd, "feed_ingest.py", &["--workspace", &workspace]).await
612 else {
613 return;
614 };
615 let intents = result
616 .get("delivery_intents")
617 .and_then(Value::as_array)
618 .cloned()
619 .unwrap_or_default();
620 if intents.is_empty() {
621 return;
622 }
623 if let Ok(delivered) = deliver_and_ack_alerts(state, &intents).await {
624 result["alerts_delivered"] = json!(delivered);
625 }
626}
627
628async fn deliver_and_ack_alerts(state: &AppState, intents: &[Value]) -> Result<usize, String> {
629 let delivered = crate::delivery::deliver_feed_intents(&state.core, intents).await?;
630 for intent in intents {
631 let Some(alert_id) = intent.get("alert_id").and_then(Value::as_i64) else {
632 continue;
633 };
634 let Some(item_id) = intent
635 .get("item")
636 .and_then(|item| item.get("id"))
637 .and_then(Value::as_i64)
638 else {
639 continue;
640 };
641 let alert_text = alert_id.to_string();
642 let item_text = item_id.to_string();
643 let result = run_feed_admin_script(
644 state.core.cwd(),
645 "feed_ingest.py",
646 &["--mark-alert-delivered", &alert_text, &item_text],
647 )
648 .await
649 .map_err(|(_, error)| error)?;
650 if result.get("status").and_then(Value::as_str) != Some("ok") {
651 return Err(format!(
652 "could not acknowledge alert {alert_id} for item {item_id}"
653 ));
654 }
655 }
656 Ok(delivered)
657}
658
659/// POST /feeds/sources — Add a new source to feeds.toml.
660pub async fn add_feed_source(
661 State(state): State<AppState>,
662 Json(payload): Json<Value>,
663) -> Result<impl IntoResponse, (StatusCode, String)> {
664 let cwd = state.core.cwd();
665 let scope = payload
666 .get("scope")
667 .and_then(Value::as_str)
668 .unwrap_or("workspace");
669 authorize_feed_mutation(&state, scope)?;
670 let config_path = if scope == "global" {
671 global_feeds_config_path()
672 } else {
673 feeds_config_path(cwd)
674 };
675
676 // Extract fields from payload
677 let name = payload
678 .get("name")
679 .and_then(|v| v.as_str())
680 .unwrap_or("Untitled");
681 let source_type = payload
682 .get("type")
683 .and_then(|v| v.as_str())
684 .unwrap_or("rss");
685 let url = payload.get("url").and_then(|v| v.as_str()).unwrap_or("");
686 if !url.is_empty() {
687 validate_source_url(url).map_err(|error| (StatusCode::BAD_REQUEST, error))?;
688 }
689 let interval = payload
690 .get("interval")
691 .and_then(|v| v.as_str())
692 .unwrap_or("1h");
693 let tags = payload.get("tags").and_then(|v| v.as_array()).map(|a| {
694 a.iter()
695 .filter_map(|v| v.as_str())
696 .collect::<Vec<_>>()
697 .join(", ")
698 });
699 let trust = payload
700 .get("trust")
701 .and_then(|v| v.as_str())
702 .unwrap_or("medium");
703
704 let source_id = payload
705 .get("id")
706 .and_then(Value::as_str)
707 .filter(|id| !id.trim().is_empty())
708 .map(ToOwned::to_owned)
709 .unwrap_or_else(|| format!("src-{}", uuid::Uuid::now_v7().simple()));
710
711 // Build the new source block
712 let mut source_block = format!(
713 "\n[[sources]]\nid = \"{}\"\nname = \"{}\"\ntype = \"{}\"\n",
714 escape_toml(&source_id),
715 escape_toml(name),
716 escape_toml(source_type)
717 );
718 if !url.is_empty() {
719 source_block.push_str(&format!("url = \"{}\"\n", escape_toml(url)));
720 }
721 source_block.push_str(&format!("interval = \"{}\"\n", escape_toml(interval)));
722 if let Some(tags_str) = &tags
723 && !tags_str.is_empty()
724 {
725 source_block.push_str(&format!(
726 "tags = [{}]\n",
727 tags_str
728 .split(", ")
729 .map(|t| format!("\"{}\"", escape_toml(t)))
730 .collect::<Vec<_>>()
731 .join(", ")
732 ));
733 }
734 source_block.push_str(&format!("trust = \"{}\"\n", escape_toml(trust)));
735 source_block.push_str("enabled = true\n");
736
737 let source_id_line = format!("id = \"{}\"", escape_toml(&source_id));
738 let conflict = format!("source id '{source_id}' already exists in {scope} scope");
739 edit_feeds_config(config_path.clone(), move |current| {
740 let mut content = current.map_or_else(default_feeds_config, ToOwned::to_owned);
741 if content.lines().any(|line| line.trim() == source_id_line) {
742 return Err((StatusCode::CONFLICT, conflict));
743 }
744 append_block(&mut content, &source_block);
745 Ok((Some(content), ()))
746 })
747 .await?;
748
749 Ok(Json(json!({
750 "status": "ok",
751 "message": format!("Source '{}' added to {}", name, config_path.display()),
752 "source": payload
753 })))
754}
755
756/// Escape a string for TOML output.
757fn escape_toml(s: &str) -> String {
758 s.replace('\\', "\\\\")
759 .replace('"', "\\\"")
760 .replace('\n', "\\n")
761}
762
763/// Find the line range `[start, end)` of a `[[table]]` array-of-tables block
764/// whose `name = "..."` field matches `name`. `end` is the index of the first
765/// line after the block: the next `[[...]]` header, a `[section]` header that
766/// isn't a dotted sub-table of this array item (e.g. `[alerts.match]` stays
767/// part of the `[[alerts]]` block it follows), or EOF.
768fn find_array_table_block(lines: &[&str], table: &str, name: &str) -> Option<(usize, usize)> {
769 let header = format!("[[{}]]", table);
770 let subtable_prefix = format!("[{}.", table);
771 let mut i = 0;
772 while i < lines.len() {
773 if lines[i].trim() == header {
774 let start = i;
775 let mut j = i + 1;
776 let mut block_key: Option<String> = None;
777 while j < lines.len() {
778 let trimmed = lines[j].trim();
779 let is_boundary = trimmed.starts_with("[[")
780 || (trimmed.starts_with('[') && !trimmed.starts_with(&subtable_prefix));
781 if is_boundary {
782 break;
783 }
784 if let Some(rest) = trimmed.strip_prefix("id") {
785 let rest = rest.trim_start();
786 if let Some(rest) = rest.strip_prefix('=') {
787 block_key = Some(unescape_toml(rest.trim().trim_matches('"')));
788 }
789 } else if block_key.is_none()
790 && let Some(rest) = trimmed.strip_prefix("name")
791 {
792 let rest = rest.trim_start();
793 if let Some(rest) = rest.strip_prefix('=') {
794 let value = rest.trim().trim_matches('"');
795 block_key = Some(unescape_toml(value));
796 }
797 }
798 j += 1;
799 }
800 if block_key.as_deref() == Some(name) {
801 return Some((start, j));
802 }
803 i = j;
804 } else {
805 i += 1;
806 }
807 }
808 None
809}
810
811/// Reverse of `escape_toml` for simple double-quoted TOML string values.
812fn unescape_toml(s: &str) -> String {
813 s.replace("\\\"", "\"").replace("\\\\", "\\")
814}
815
816/// Remove a `[[table]]` block by name from TOML `content`. Returns the new
817/// content, or `None` if no matching block was found.
818fn remove_array_table_block(content: &str, table: &str, name: &str) -> Option<String> {
819 let lines: Vec<&str> = content.lines().collect();
820 let (start, end) = find_array_table_block(&lines, table, name)?;
821 let mut new_lines: Vec<&str> = Vec::with_capacity(lines.len());
822 new_lines.extend_from_slice(&lines[..start]);
823 new_lines.extend_from_slice(&lines[end..]);
824 Some(new_lines.join("\n"))
825}
826
827/// Set (or add) a scalar `field = value` line inside a `[[table]]` block
828/// matched by name, replacing the existing line for that field if present.
829fn set_field_in_block(
830 content: &str,
831 table: &str,
832 name: &str,
833 field: &str,
834 value_line: &str,
835) -> Option<String> {
836 let lines: Vec<&str> = content.lines().collect();
837 let (start, end) = find_array_table_block(&lines, table, name)?;
838 let mut new_lines: Vec<String> = lines[..=start].iter().map(|l| l.to_string()).collect();
839 let mut replaced = false;
840 for line in &lines[start + 1..end] {
841 let trimmed = line.trim();
842 if trimmed.starts_with(field) && trimmed[field.len()..].trim_start().starts_with('=') {
843 new_lines.push(value_line.to_string());
844 replaced = true;
845 } else {
846 new_lines.push(line.to_string());
847 }
848 }
849 if !replaced {
850 new_lines.push(value_line.to_string());
851 }
852 new_lines.extend(lines[end..].iter().map(|l| l.to_string()));
853 Some(new_lines.join("\n"))
854}
855
856fn default_feeds_config() -> String {
857 "[general]\ndefault_check_interval = \"30m\"\nmax_items_per_feed = 500\ndedup_window_days = 90\n\n".to_string()
858}
859
860fn append_block(content: &mut String, block: &str) {
861 if !content.is_empty() && !content.ends_with('\n') {
862 content.push('\n');
863 }
864 content.push_str(block);
865}
866
867/// Every edit to a `feeds.toml` goes through here: `edit` sees the current
868/// text (`None` when the file does not exist) and returns the new text, or
869/// `None` to leave it alone. The read, the edit and the atomic replace hold
870/// one lock (`vak_config::file_update`), so concurrent edits never publish a
871/// half-written file or rewrite the file from a stale read.
872async fn edit_feeds_config<T: Send + 'static>(
873 config_path: PathBuf,
874 edit: impl FnOnce(Option<&str>) -> Result<(Option<String>, T), (StatusCode, String)>
875 + Send
876 + 'static,
877) -> Result<T, (StatusCode, String)> {
878 tokio::task::spawn_blocking(move || {
879 vak_config::file_update::update_file(&config_path, edit).map_err(|error| match error {
880 vak_config::file_update::UpdateError::Edit(error) => error,
881 vak_config::file_update::UpdateError::Io { path, source } => (
882 StatusCode::INTERNAL_SERVER_ERROR,
883 format!("Failed to update {}: {source}", path.display()),
884 ),
885 })
886 })
887 .await
888 .map_err(|error| {
889 (
890 StatusCode::INTERNAL_SERVER_ERROR,
891 format!("Feed config update did not finish: {error}"),
892 )
893 })?
894}
895
896/// DELETE /feeds/sources/{name} — Remove a feed source from config.
897/// DELETE /feeds/sources/{name} — Remove a feed source.
898///
899/// Soft-deletes the materialized source row and removes its declaration from
900/// the canonical scope-specific configuration.
901pub async fn delete_feed_source(
902 Path(source_id): Path<String>,
903 Query(params): Query<HashMap<String, String>>,
904 State(state): State<AppState>,
905) -> Result<impl IntoResponse, (StatusCode, String)> {
906 let cwd = state.core.cwd();
907 let scope = params
908 .get("scope")
909 .map(String::as_str)
910 .unwrap_or("workspace");
911 authorize_feed_mutation(&state, scope)?;
912
913 let removed = run_feed_admin_script(
914 cwd,
915 "feed_ingest.py",
916 &["--remove-source", &source_id, "--scope", scope],
917 )
918 .await?
919 .get("status")
920 .and_then(|s| s.as_str())
921 == Some("ok");
922
923 if !removed {
924 return Err((
925 StatusCode::NOT_FOUND,
926 format!("Source '{}' not found", source_id),
927 ));
928 }
929
930 let config_path = if scope == "global" {
931 global_feeds_config_path()
932 } else {
933 feeds_config_path(cwd)
934 };
935 let declared = source_id.clone();
936 let _ = edit_feeds_config(config_path, move |current| {
937 let next =
938 current.and_then(|content| remove_array_table_block(content, "sources", &declared));
939 Ok((next, ()))
940 })
941 .await;
942
943 Ok(Json(json!({
944 "status": "ok",
945 "message": format!("Source '{}' removed", source_id),
946 })))
947}
948
949/// PATCH /feeds/sources/{name} — Update a source's enabled/interval/trust fields.
950///
951/// `tags` is **refused** here. The `feeds` table has no tags column, so a
952/// tag written on update reached feeds.toml and nothing else: the DB never
953/// saw it and the UI never showed it. A PATCH that silently half-applies
954/// is worse than one that says no, so the caller is told where tags are
955/// actually set instead (AGENTS.md invariant 30).
956pub async fn update_feed_source(
957 Path(source_id): Path<String>,
958 Query(params): Query<HashMap<String, String>>,
959 State(state): State<AppState>,
960 Json(payload): Json<Value>,
961) -> Result<impl IntoResponse, (StatusCode, String)> {
962 let cwd = state.core.cwd();
963 let scope = params
964 .get("scope")
965 .map(String::as_str)
966 .unwrap_or("workspace");
967 authorize_feed_mutation(&state, scope)?;
968
969 let enabled = payload.get("enabled").and_then(|v| v.as_bool());
970 let interval = payload.get("interval").and_then(|v| v.as_str());
971 let trust = payload.get("trust").and_then(|v| v.as_str());
972 if payload.get("tags").is_some() {
973 return Err((
974 StatusCode::BAD_REQUEST,
975 "tags cannot be updated here: the feeds table has no tags column, so the \
976 change would reach feeds.toml and nothing else. Set tags when creating \
977 the source."
978 .into(),
979 ));
980 }
981
982 if enabled.is_none() && interval.is_none() && trust.is_none() {
983 return Err((
984 StatusCode::BAD_REQUEST,
985 "No recognized fields to update (enabled, interval, trust)".into(),
986 ));
987 }
988
989 if enabled.is_some() || interval.is_some() || trust.is_some() {
990 let mut args: Vec<String> = vec!["--update-source".into(), source_id.clone()];
991 args.extend(["--scope".into(), scope.into()]);
992 if let Some(e) = enabled {
993 args.push("--set-enabled".into());
994 args.push(if e { "true" } else { "false" }.into());
995 }
996 if let Some(i) = interval {
997 args.push("--set-interval".into());
998 args.push(i.into());
999 }
1000 if let Some(t) = trust {
1001 args.push("--set-trust".into());
1002 args.push(t.into());
1003 }
1004 let arg_refs: Vec<&str> = args.iter().map(|s| s.as_str()).collect();
1005 let updated = run_feed_admin_script(cwd, "feed_ingest.py", &arg_refs)
1006 .await?
1007 .get("status")
1008 .and_then(|s| s.as_str())
1009 == Some("ok");
1010 if !updated {
1011 return Err((
1012 StatusCode::NOT_FOUND,
1013 format!("Source '{}' not found", source_id),
1014 ));
1015 }
1016 }
1017
1018 // Reconcile the durable declaration when this source has one. The
1019 // materialized row above remains authoritative for the current response.
1020 let config_path = if scope == "global" {
1021 global_feeds_config_path()
1022 } else {
1023 feeds_config_path(cwd)
1024 };
1025 let mut fields = Vec::new();
1026 if let Some(e) = enabled {
1027 fields.push(("enabled", format!("enabled = {e}")));
1028 }
1029 if let Some(i) = interval {
1030 fields.push(("interval", format!("interval = \"{}\"", escape_toml(i))));
1031 }
1032 if let Some(t) = trust {
1033 fields.push(("trust", format!("trust = \"{}\"", escape_toml(t))));
1034 }
1035 let declared = source_id.clone();
1036 let _ = edit_feeds_config(config_path, move |current| {
1037 let Some(mut content) = current.map(ToOwned::to_owned) else {
1038 return Ok((None, ()));
1039 };
1040 let mut touched = false;
1041 for (field, line) in &fields {
1042 if let Some(next) = set_field_in_block(&content, "sources", &declared, field, line) {
1043 content = next;
1044 touched = true;
1045 }
1046 }
1047 Ok((touched.then_some(content), ()))
1048 })
1049 .await;
1050
1051 Ok(Json(json!({
1052 "status": "ok",
1053 "message": format!("Source '{}' updated", source_id),
1054 })))
1055}
1056
1057/// GET /feeds/sources/configured — List sources actually known to the DB
1058/// (populated by ingestion), with live enabled/trust/interval status.
1059pub async fn list_configured_sources(
1060 State(state): State<AppState>,
1061) -> Result<impl IntoResponse, (StatusCode, String)> {
1062 let cwd = state.core.cwd();
1063 let request = json!({
1064 "jsonrpc": "2.0",
1065 "id": 1,
1066 "method": "tools/call",
1067 "params": {
1068 "name": "feed_sources",
1069 "arguments": {}
1070 }
1071 });
1072 let result = run_feed_mcp_request(cwd, &request).await?;
1073 Ok(Json(result))
1074}
1075
1076/// POST /feeds/alerts — Add a new alert rule to feeds.toml.
1077pub async fn add_feed_alert(
1078 State(state): State<AppState>,
1079 Json(payload): Json<Value>,
1080) -> Result<impl IntoResponse, (StatusCode, String)> {
1081 let cwd = state.core.cwd();
1082 let scope = payload
1083 .get("scope")
1084 .and_then(Value::as_str)
1085 .unwrap_or("workspace");
1086 authorize_feed_mutation(&state, scope)?;
1087 let config_path = if scope == "global" {
1088 global_feeds_config_path()
1089 } else {
1090 feeds_config_path(cwd)
1091 };
1092
1093 let name = payload
1094 .get("name")
1095 .and_then(|v| v.as_str())
1096 .filter(|s| !s.is_empty())
1097 .ok_or((StatusCode::BAD_REQUEST, "Alert 'name' is required".into()))?;
1098
1099 let str_array = |key: &str| -> Vec<String> {
1100 payload
1101 .get(key)
1102 .and_then(|v| v.as_array())
1103 .map(|a| {
1104 a.iter()
1105 .filter_map(|v| v.as_str())
1106 .map(String::from)
1107 .collect()
1108 })
1109 .unwrap_or_default()
1110 };
1111 let keywords = str_array("keywords");
1112 let tags = str_array("tags");
1113 let sources = str_array("sources");
1114
1115 if keywords.is_empty() && tags.is_empty() && sources.is_empty() {
1116 return Err((
1117 StatusCode::BAD_REQUEST,
1118 "At least one of keywords, tags, or sources must be set".into(),
1119 ));
1120 }
1121
1122 let action = payload
1123 .get("action")
1124 .and_then(|v| v.as_str())
1125 .unwrap_or("deliver");
1126 if !matches!(action, "deliver" | "hook" | "both") {
1127 return Err((
1128 StatusCode::BAD_REQUEST,
1129 "action must be one of: deliver, hook, both".into(),
1130 ));
1131 }
1132 let deliver_to = payload
1133 .get("deliver_to")
1134 .and_then(|v| v.as_str())
1135 .unwrap_or("");
1136 if matches!(action, "deliver" | "both") && deliver_to.trim().is_empty() {
1137 return Err((
1138 StatusCode::BAD_REQUEST,
1139 "deliver_to is required for deliver and both alerts".into(),
1140 ));
1141 }
1142 let cooldown_minutes = payload
1143 .get("cooldown_minutes")
1144 .and_then(|v| v.as_i64())
1145 .unwrap_or(30);
1146
1147 let to_toml_array = |items: &[String]| -> String {
1148 items
1149 .iter()
1150 .map(|t| format!("\"{}\"", escape_toml(t)))
1151 .collect::<Vec<_>>()
1152 .join(", ")
1153 };
1154
1155 let mut block = format!(
1156 "\n[[alerts]]\nname = \"{}\"\naction = \"{}\"\ncooldown_minutes = {}\nenabled = true\n",
1157 escape_toml(name),
1158 escape_toml(action),
1159 cooldown_minutes
1160 );
1161 if !deliver_to.is_empty() {
1162 block.push_str(&format!("deliver_to = \"{}\"\n", escape_toml(deliver_to)));
1163 }
1164 block.push_str("\n[alerts.match]\n");
1165 if !keywords.is_empty() {
1166 block.push_str(&format!("keywords = [{}]\n", to_toml_array(&keywords)));
1167 }
1168 if !tags.is_empty() {
1169 block.push_str(&format!("tags = [{}]\n", to_toml_array(&tags)));
1170 }
1171 if !sources.is_empty() {
1172 block.push_str(&format!("sources = [{}]\n", to_toml_array(&sources)));
1173 }
1174
1175 edit_feeds_config(config_path.clone(), move |current| {
1176 let mut content = current.map_or_else(default_feeds_config, ToOwned::to_owned);
1177 append_block(&mut content, &block);
1178 Ok((Some(content), ()))
1179 })
1180 .await?;
1181 run_feed_script(cwd, "feed_ingest.py", &["--sync-alerts"]).await?;
1182
1183 Ok(Json(json!({
1184 "status": "ok",
1185 "message": format!("Alert '{}' added to {}", name, config_path.display()),
1186 })))
1187}
1188
1189/// DELETE /feeds/alerts/{name} — Remove an alert rule from config.
1190pub async fn delete_feed_alert(
1191 Path(name): Path<String>,
1192 Query(params): Query<HashMap<String, String>>,
1193 State(state): State<AppState>,
1194) -> Result<impl IntoResponse, (StatusCode, String)> {
1195 let cwd = state.core.cwd();
1196 let scope = params
1197 .get("scope")
1198 .map(String::as_str)
1199 .unwrap_or("workspace");
1200 authorize_feed_mutation(&state, scope)?;
1201 let config_path = if scope == "global" {
1202 global_feeds_config_path()
1203 } else {
1204 feeds_config_path(cwd)
1205 };
1206
1207 let alert = name.clone();
1208 edit_feeds_config(config_path.clone(), move |current| {
1209 let content =
1210 current.ok_or_else(|| (StatusCode::NOT_FOUND, "Config file not found".to_string()))?;
1211 let next = remove_array_table_block(content, "alerts", &alert)
1212 .ok_or_else(|| (StatusCode::NOT_FOUND, format!("Alert '{alert}' not found")))?;
1213 Ok((Some(next), ()))
1214 })
1215 .await?;
1216 run_feed_script(cwd, "feed_ingest.py", &["--sync-alerts"]).await?;
1217
1218 Ok(Json(json!({
1219 "status": "ok",
1220 "message": format!("Alert '{}' removed from {}", name, config_path.display()),
1221 })))
1222}
1223
1224/// Build feed routes.
1225pub fn routes() -> Router<AppState> {
1226 Router::new()
1227 .route(
1228 "/feeds/sources",
1229 get(list_source_types).post(add_feed_source),
1230 )
1231 .route("/feeds/sources/configured", get(list_configured_sources))
1232 .route(
1233 "/feeds/sources/{name}",
1234 delete(delete_feed_source).patch(update_feed_source),
1235 )
1236 .route("/feeds/config", get(get_feed_config))
1237 .route("/feeds/items", get(list_feed_items))
1238 .route("/feeds/items/{id}", get(get_feed_item))
1239 .route("/feeds/search", get(search_feed_items))
1240 .route("/feeds/stats", get(get_feed_stats))
1241 .route("/feeds/alerts", get(get_feed_alerts).post(add_feed_alert))
1242 .route("/feeds/runs", get(get_feed_runs))
1243 .route("/feeds/quarantine", get(get_feed_quarantine))
1244 .route("/feeds/quarantine/{id}/release", post(release_feed_item))
1245 .route("/feeds/alerts/{name}", delete(delete_feed_alert))
1246 .route("/feeds/ingest", post(trigger_ingestion))
1247}
1248
1249#[cfg(test)]
1250#[allow(clippy::panic, clippy::unwrap_used, clippy::expect_used)]
1251mod tests {
1252 use super::{authorize_feed_scope, feed_environment, feeds_dir, validate_source_url};
1253 use vak_config::PermissionMode;
1254
1255 /// The feed subprocess runs with `env_clear`, so `feed_environment` is the
1256 /// complete list of what it sees. It must carry the operational minimum
1257 /// (PATH, so `python3` resolves) and never a parent secret — before this,
1258 /// the subprocess inherited the whole server environment, `VAK_GATEWAY_TOKEN`
1259 /// and provider keys included (invariant 12).
1260 #[test]
1261 fn feed_subprocess_environment_is_a_secret_free_allowlist() {
1262 let _home = vak_config::paths::isolate_home_for_tests();
1263 let environment = feed_environment(
1264 std::path::Path::new("/tmp/ws"),
1265 std::path::Path::new("/tmp/scripts"),
1266 );
1267 let allowed = [
1268 "PYTHONPATH",
1269 "VAK_FEED_WORKSPACE",
1270 "VAK_FEEDS_DB",
1271 "VAK_FEEDS_LOG",
1272 "VAK_FEEDS_CONFIG",
1273 "PATH",
1274 "HOME",
1275 ];
1276 for (key, _) in &environment {
1277 assert!(allowed.contains(key), "{key} is not in the feed allowlist");
1278 }
1279 assert!(
1280 environment.iter().any(|(key, _)| *key == "PATH"),
1281 "PATH must be passed so python3 resolves under env_clear"
1282 );
1283 for secret in ["VAK_GATEWAY_TOKEN", "ANTHROPIC_API_KEY", "OPENAI_API_KEY"] {
1284 assert!(
1285 !environment.iter().any(|(key, _)| *key == secret),
1286 "{secret} must never be passed to a feed subprocess"
1287 );
1288 }
1289 }
1290
1291 /// The script the server runs is located only from the binary, never from
1292 /// the session workspace: a workspace with its own `scripts/feeds` cannot
1293 /// supply the code the server executes (invariants 12, 14, 15).
1294 #[test]
1295 fn feeds_dir_never_resolves_from_a_workspace() {
1296 let workspace = tempfile::tempdir().expect("tempdir");
1297 let planted = workspace.path().join("scripts").join("feeds");
1298 std::fs::create_dir_all(&planted).expect("mkdir");
1299 std::fs::write(planted.join("feed_ingest.py"), b"raise SystemExit\n").expect("write");
1300 assert!(
1301 !feeds_dir().starts_with(workspace.path()),
1302 "feeds_dir resolved a script from the session workspace"
1303 );
1304 }
1305
1306 /// Every path a feed script writes comes from the canonical data home,
1307 /// so an overridden `VAK_HOME` holds the feed store, its log and its
1308 /// config; the scripts used to work out the platform default for
1309 /// themselves and wrote outside it.
1310 #[test]
1311 fn feeds_write_under_overridden_home() {
1312 let home = vak_config::paths::isolate_home_for_tests();
1313 let environment = feed_environment(
1314 std::path::Path::new("/tmp/ws"),
1315 std::path::Path::new("/tmp/scripts"),
1316 );
1317 for name in ["VAK_FEEDS_DB", "VAK_FEEDS_LOG", "VAK_FEEDS_CONFIG"] {
1318 let (_, value) = environment
1319 .iter()
1320 .find(|(key, _)| *key == name)
1321 .unwrap_or_else(|| panic!("{name} is passed"));
1322 assert!(
1323 std::path::Path::new(value).starts_with(&home),
1324 "{name} = {value} is outside {}",
1325 home.display()
1326 );

Feeds

Indexing the workspace…

Feeds