mirror of
https://github.com/fabro-sh/fabro.git
synced 2026-10-07 03:00:29 +00:00
Merge remote-tracking branch 'origin/main' into codex/glm-5-2-support
# Conflicts: # lib/crates/fabro-llm/tests/integration.rs # lib/crates/fabro-model/src/catalog.rs
This commit is contained in:
commit
8a243085d0
13 changed files with 468 additions and 29 deletions
|
|
@ -55,8 +55,8 @@ pub use question_tools::{
|
|||
};
|
||||
pub use read_before_write_sandbox::ReadBeforeWriteSandbox;
|
||||
pub use sandbox::{
|
||||
CommandOutputCallback, DirEntry, ExecResult, ExecStreamingResult, GrepOptions, Sandbox,
|
||||
SandboxEvent, SandboxEventCallback, StderrCollector, StdioProcess, StdioProcessHandle,
|
||||
CommandOutputCallback, DirEntry, ExecResult, ExecStreamingResult, GrepOptions, RefreshOutcome,
|
||||
Sandbox, SandboxEvent, SandboxEventCallback, StderrCollector, StdioProcess, StdioProcessHandle,
|
||||
WorktreeEvent, WorktreeEventCallback, WorktreeOptions, WorktreeSandbox, format_lines_numbered,
|
||||
shell_quote,
|
||||
};
|
||||
|
|
|
|||
|
|
@ -2,8 +2,8 @@
|
|||
// Re-export the delegate_sandbox! macro at crate root so existing
|
||||
// `crate::delegate_sandbox!` invocations continue to work.
|
||||
pub use fabro_sandbox::{
|
||||
CommandOutputCallback, DirEntry, ExecResult, ExecStreamingResult, GrepOptions, Sandbox,
|
||||
SandboxEvent, SandboxEventCallback, StderrCollector, StdioProcess, StdioProcessHandle,
|
||||
CommandOutputCallback, DirEntry, ExecResult, ExecStreamingResult, GrepOptions, RefreshOutcome,
|
||||
Sandbox, SandboxEvent, SandboxEventCallback, StderrCollector, StdioProcess, StdioProcessHandle,
|
||||
StdioProcessTermination, WorktreeEvent, WorktreeEventCallback, WorktreeOptions,
|
||||
WorktreeSandbox, delegate_sandbox, format_lines_numbered, shell_quote,
|
||||
};
|
||||
|
|
|
|||
|
|
@ -5,7 +5,10 @@
|
|||
|
||||
use std::sync::Arc;
|
||||
|
||||
use fabro_auth::ApiCredential;
|
||||
use fabro_llm::client::Client;
|
||||
use fabro_llm::error::ProviderErrorKind;
|
||||
use fabro_llm::model_test::{ModelTestStatus, run_model_test};
|
||||
use fabro_llm::provider::ProviderAdapter;
|
||||
use fabro_llm::providers::{
|
||||
AnthropicAdapter, BedrockAdapter, GeminiAdapter, OpenAiAdapter, OpenAiCompatibleAdapter,
|
||||
|
|
@ -13,8 +16,8 @@ use fabro_llm::providers::{
|
|||
use fabro_llm::types::{
|
||||
CostSource, FinishReason, Message, ReasoningEffort, Request, ToolChoice, ToolDefinition,
|
||||
};
|
||||
use fabro_model::Catalog;
|
||||
use fabro_model::catalog::LlmCatalogSettings;
|
||||
use fabro_model::catalog::{LlmCatalogSettings, ProviderCatalogSettings};
|
||||
use fabro_model::{Catalog, ModelTestMode, ProviderId};
|
||||
use fabro_static::EnvVars;
|
||||
|
||||
fn make_request(model: &str) -> Request {
|
||||
|
|
@ -497,6 +500,42 @@ enabled = true
|
|||
assert_eq!(final_response.cost_source, Some(CostSource::Authoritative));
|
||||
}
|
||||
|
||||
#[fabro_macros::e2e_test(live("OPENROUTER_API_KEY"))]
|
||||
async fn openrouter_kimi_k3_deep_tool_round_trip() {
|
||||
let api_key =
|
||||
std::env::var(EnvVars::OPENROUTER_API_KEY).expect("OPENROUTER_API_KEY must be set");
|
||||
let provider = ProviderId::new("openrouter");
|
||||
let mut settings = LlmCatalogSettings::default();
|
||||
settings
|
||||
.providers
|
||||
.insert(provider.to_string(), ProviderCatalogSettings {
|
||||
enabled: Some(true),
|
||||
..ProviderCatalogSettings::default()
|
||||
});
|
||||
let catalog = Arc::new(
|
||||
Catalog::from_builtin_with_overrides(&settings)
|
||||
.expect("enabled OpenRouter catalog should build"),
|
||||
);
|
||||
let credential = ApiCredential::from_api_key(provider, api_key, &catalog)
|
||||
.expect("OpenRouter credential should resolve from the catalog");
|
||||
let client = Arc::new(
|
||||
Client::from_credentials(vec![credential], Arc::clone(&catalog))
|
||||
.await
|
||||
.expect("OpenRouter client should build from the catalog"),
|
||||
);
|
||||
let model = catalog
|
||||
.get("moonshotai/kimi-k3")
|
||||
.expect("OpenRouter Kimi K3 should be present");
|
||||
|
||||
let outcome = run_model_test(model, ModelTestMode::Deep, client).await;
|
||||
assert_eq!(
|
||||
outcome.status,
|
||||
ModelTestStatus::Ok,
|
||||
"OpenRouter Kimi K3 deep test failed: {:?}",
|
||||
outcome.error_message
|
||||
);
|
||||
}
|
||||
|
||||
async fn run_multi_turn_cache_test(
|
||||
adapter: &dyn ProviderAdapter,
|
||||
model: &str,
|
||||
|
|
|
|||
|
|
@ -2120,6 +2120,71 @@ enabled = true
|
|||
]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn builtin_openrouter_includes_kimi_k3_when_enabled() {
|
||||
let catalog = Catalog::from_builtin_with_overrides(&minimal_settings(
|
||||
r"
|
||||
[providers.openrouter]
|
||||
enabled = true
|
||||
",
|
||||
))
|
||||
.expect("enabled OpenRouter override should build from the built-in provider settings");
|
||||
|
||||
let model = catalog
|
||||
.get("moonshotai/kimi-k3")
|
||||
.expect("OpenRouter Kimi K3 should be present");
|
||||
insta::assert_debug_snapshot!(model, @r#"
|
||||
Model {
|
||||
id: "moonshotai/kimi-k3",
|
||||
provider: openrouter,
|
||||
family: "kimi-k3",
|
||||
display_name: "Kimi K3 (via OpenRouter)",
|
||||
limits: ModelLimits {
|
||||
context_window: 1048576,
|
||||
max_output: Some(
|
||||
131072,
|
||||
),
|
||||
},
|
||||
training: None,
|
||||
knowledge_cutoff: None,
|
||||
features: ModelFeatures {
|
||||
tools: true,
|
||||
vision: true,
|
||||
reasoning: true,
|
||||
reasoning_effort: AlwaysAdaptive,
|
||||
prompt_cache: true,
|
||||
sampling_params: false,
|
||||
},
|
||||
costs: ModelCosts {
|
||||
input_cost_per_mtok: Some(
|
||||
3.0,
|
||||
),
|
||||
output_cost_per_mtok: Some(
|
||||
15.0,
|
||||
),
|
||||
cache_input_cost_per_mtok: Some(
|
||||
0.3,
|
||||
),
|
||||
},
|
||||
estimated_output_tps: None,
|
||||
aliases: [],
|
||||
default: false,
|
||||
small_default: false,
|
||||
configured: false,
|
||||
}
|
||||
"#);
|
||||
|
||||
let settings = catalog
|
||||
.model_settings("moonshotai/kimi-k3")
|
||||
.expect("OpenRouter Kimi K3 settings should be present");
|
||||
assert_eq!(settings.api_id, "moonshotai/kimi-k3");
|
||||
assert_eq!(settings.controls.reasoning_effort, vec![
|
||||
ReasoningEffort::Low,
|
||||
ReasoningEffort::High,
|
||||
ReasoningEffort::Max,
|
||||
]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn builtin_ollama_provider_is_opt_in() {
|
||||
let ollama = ProviderId::new("ollama");
|
||||
|
|
|
|||
|
|
@ -278,6 +278,32 @@ reasoning = false
|
|||
input_cost_per_mtok = 0.73
|
||||
output_cost_per_mtok = 3.49
|
||||
|
||||
[models."moonshotai/kimi-k3"]
|
||||
provider = "openrouter"
|
||||
api_id = "moonshotai/kimi-k3"
|
||||
display_name = "Kimi K3 (via OpenRouter)"
|
||||
family = "kimi-k3"
|
||||
|
||||
[models."moonshotai/kimi-k3".limits]
|
||||
context_window = 1048576
|
||||
max_output = 131072
|
||||
|
||||
[models."moonshotai/kimi-k3".features]
|
||||
tools = true
|
||||
vision = true
|
||||
reasoning = true
|
||||
reasoning_effort = "always_adaptive"
|
||||
prompt_cache = true
|
||||
sampling_params = false
|
||||
|
||||
[models."moonshotai/kimi-k3".controls]
|
||||
reasoning_effort = ["low", "high", "max"]
|
||||
|
||||
[models."moonshotai/kimi-k3".costs]
|
||||
input_cost_per_mtok = 3.0
|
||||
output_cost_per_mtok = 15.0
|
||||
cache_input_cost_per_mtok = 0.3
|
||||
|
||||
[models."qwen/qwen3-coder"]
|
||||
provider = "openrouter"
|
||||
api_id = "qwen/qwen3-coder"
|
||||
|
|
|
|||
|
|
@ -26,7 +26,7 @@ use tokio_util::sync::CancellationToken;
|
|||
|
||||
use crate::clone_source::{self, CloneDecision, EmptyWorkspaceReason};
|
||||
use crate::redact::redact_auth_url;
|
||||
use crate::sandbox::{optional_timeout, resolve_path};
|
||||
use crate::sandbox::{RefreshOutcome, optional_timeout, resolve_path};
|
||||
use crate::{
|
||||
CommandOutputCallback, DirEntry, ExecResult, ExecStreamingResult, GrepOptions, Sandbox,
|
||||
SandboxEvent, SandboxEventCallback, StdioProcess, glob_match, managed_labels, shell_quote,
|
||||
|
|
@ -1357,16 +1357,22 @@ impl Sandbox for DaytonaSandbox {
|
|||
Ok(Some((preview.url, headers)))
|
||||
}
|
||||
|
||||
async fn refresh_push_credentials(&self) -> crate::Result<()> {
|
||||
async fn refresh_push_credentials(&self) -> crate::Result<RefreshOutcome> {
|
||||
if !self.repo_cloned() {
|
||||
return Ok(());
|
||||
return Ok(RefreshOutcome::Skipped);
|
||||
}
|
||||
let Some(origin_url) = self.origin_url.get() else {
|
||||
return Ok(()); // no authenticated origin — nothing to refresh
|
||||
return Ok(RefreshOutcome::Skipped); // no authenticated origin — nothing to refresh
|
||||
};
|
||||
let Some(creds) = &self.github_app else {
|
||||
return Ok(());
|
||||
return Ok(RefreshOutcome::Skipped);
|
||||
};
|
||||
// Only a GitHub App installation token can be re-minted; a static PAT or
|
||||
// a pre-minted Installation token is fixed, so re-embedding it changes
|
||||
// nothing. Short-circuit to Skipped before the resolve + set-url exec.
|
||||
if !matches!(creds, GitHubCredentials::App(_)) {
|
||||
return Ok(RefreshOutcome::Skipped);
|
||||
}
|
||||
|
||||
let auth_url = fabro_github::resolve_authenticated_url(
|
||||
&fabro_github::GitHubContext::new(creds, &fabro_github::github_api_base_url()),
|
||||
|
|
@ -1394,7 +1400,9 @@ impl Sandbox for DaytonaSandbox {
|
|||
));
|
||||
}
|
||||
|
||||
Ok(())
|
||||
// Static creds were short-circuited to Skipped above; reaching here means
|
||||
// a GitHub App installation token was freshly minted.
|
||||
Ok(RefreshOutcome::Refreshed)
|
||||
}
|
||||
|
||||
async fn set_autostop_interval(&self, minutes: i32) -> crate::Result<()> {
|
||||
|
|
|
|||
|
|
@ -28,7 +28,7 @@ use tokio_util::sync::CancellationToken;
|
|||
use crate::clone_source::{self, CloneDecision, EmptyWorkspaceReason};
|
||||
use crate::managed_labels::{self, MANAGED_LABEL, RUN_ID_LABEL};
|
||||
use crate::redact::redact_auth_url;
|
||||
use crate::sandbox::{StdioProcessControl, optional_timeout, resolve_path};
|
||||
use crate::sandbox::{RefreshOutcome, StdioProcessControl, optional_timeout, resolve_path};
|
||||
use crate::{
|
||||
CommandOutputCallback, DEFAULT_EXEC_OUTPUT_TAIL_BYTES, DirEntry, ExecResult,
|
||||
ExecStreamingResult, GrepOptions, Sandbox, SandboxEvent, SandboxEventCallback, StderrCollector,
|
||||
|
|
@ -1952,16 +1952,22 @@ impl Sandbox for DockerSandbox {
|
|||
self.origin_url.get().map(String::as_str)
|
||||
}
|
||||
|
||||
async fn refresh_push_credentials(&self) -> crate::Result<()> {
|
||||
async fn refresh_push_credentials(&self) -> crate::Result<RefreshOutcome> {
|
||||
if !self.repo_cloned() {
|
||||
return Ok(());
|
||||
return Ok(RefreshOutcome::Skipped);
|
||||
}
|
||||
let Some(origin_url) = self.origin_url.get() else {
|
||||
return Ok(());
|
||||
return Ok(RefreshOutcome::Skipped);
|
||||
};
|
||||
let Some(creds) = &self.github_app else {
|
||||
return Ok(());
|
||||
return Ok(RefreshOutcome::Skipped);
|
||||
};
|
||||
// Only a GitHub App installation token can be re-minted; a static PAT or
|
||||
// a pre-minted Installation token is fixed, so re-embedding it changes
|
||||
// nothing. Short-circuit to Skipped before the resolve + set-url exec.
|
||||
if !matches!(creds, GitHubCredentials::App(_)) {
|
||||
return Ok(RefreshOutcome::Skipped);
|
||||
}
|
||||
|
||||
let auth_url = fabro_github::resolve_authenticated_url(
|
||||
&fabro_github::GitHubContext::new(creds, &fabro_github::github_api_base_url()),
|
||||
|
|
@ -1986,7 +1992,9 @@ impl Sandbox for DockerSandbox {
|
|||
));
|
||||
}
|
||||
|
||||
Ok(())
|
||||
// Static creds were short-circuited to Skipped above; reaching here means
|
||||
// a GitHub App installation token was freshly minted.
|
||||
Ok(RefreshOutcome::Refreshed)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -55,8 +55,8 @@ pub use read_guard::ReadBeforeWriteSandbox;
|
|||
pub use reconnect::{reconnect, reconnect_for_run, reconnect_for_run_with_callback};
|
||||
pub use sandbox::{
|
||||
CommandOutputCallback, DEFAULT_EXEC_OUTPUT_TAIL_BYTES, DirEntry, ExecResult,
|
||||
ExecStreamingResult, GitRunInfo, GitSetupIntent, GrepOptions, Sandbox, SandboxEvent,
|
||||
SandboxEventCallback, StderrCollector, StdioProcess, StdioProcessHandle,
|
||||
ExecStreamingResult, GitRunInfo, GitSetupIntent, GrepOptions, RefreshOutcome, Sandbox,
|
||||
SandboxEvent, SandboxEventCallback, StderrCollector, StdioProcess, StdioProcessHandle,
|
||||
StdioProcessTermination, format_lines_numbered, git_push_via_exec, redacted_output_tail,
|
||||
setup_git_via_exec, shell_quote,
|
||||
};
|
||||
|
|
|
|||
|
|
@ -197,7 +197,7 @@ macro_rules! delegate_sandbox {
|
|||
self.$field.snapshot_info()
|
||||
}
|
||||
|
||||
async fn refresh_push_credentials(&self) -> $crate::Result<()> {
|
||||
async fn refresh_push_credentials(&self) -> $crate::Result<$crate::RefreshOutcome> {
|
||||
self.$field.refresh_push_credentials().await
|
||||
}
|
||||
|
||||
|
|
@ -812,6 +812,18 @@ pub struct GrepOptions {
|
|||
pub max_results: Option<usize>,
|
||||
}
|
||||
|
||||
/// Outcome of [`Sandbox::refresh_push_credentials`]: whether a fresh token was
|
||||
/// actually minted and applied to the origin remote, or the call was a no-op
|
||||
/// (no clone, no authenticated origin, or no GitHub App credentials to rotate).
|
||||
/// Lets callers log accurately instead of assuming every `Ok` re-minted.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum RefreshOutcome {
|
||||
/// A fresh token was minted and the origin remote URL was updated.
|
||||
Refreshed,
|
||||
/// Nothing to refresh (no clone / no origin / no managed credentials).
|
||||
Skipped,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
pub trait Sandbox: Send + Sync {
|
||||
async fn read_file_bytes(&self, path: &str) -> crate::Result<Vec<u8>>;
|
||||
|
|
@ -960,10 +972,11 @@ pub trait Sandbox: Send + Sync {
|
|||
}
|
||||
|
||||
/// Refresh git push credentials (e.g. rotate an expiring GitHub App token).
|
||||
/// Default is a no-op; Daytona overrides to update the remote URL with a
|
||||
/// fresh token.
|
||||
async fn refresh_push_credentials(&self) -> crate::Result<()> {
|
||||
Ok(())
|
||||
/// Default is a no-op; Docker/Daytona override to update the remote URL
|
||||
/// with a fresh token. Returns [`RefreshOutcome`] so callers can tell
|
||||
/// an actual re-mint from a skipped no-op.
|
||||
async fn refresh_push_credentials(&self) -> crate::Result<RefreshOutcome> {
|
||||
Ok(RefreshOutcome::Skipped)
|
||||
}
|
||||
|
||||
/// Set the auto-stop interval in minutes (0 to disable).
|
||||
|
|
|
|||
|
|
@ -361,7 +361,7 @@ impl Sandbox for WorktreeSandbox {
|
|||
self.inner.sandbox_info()
|
||||
}
|
||||
|
||||
async fn refresh_push_credentials(&self) -> crate::Result<()> {
|
||||
async fn refresh_push_credentials(&self) -> crate::Result<crate::RefreshOutcome> {
|
||||
self.inner.refresh_push_credentials().await
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -13,6 +13,11 @@ const WORKER_ENV_ALLOWLIST: &[&str] = &[
|
|||
EnvVars::FABRO_LOG,
|
||||
EnvVars::FABRO_HOME,
|
||||
EnvVars::FABRO_STORAGE_ROOT,
|
||||
// Push-credential refresh-ahead tunables (FABRO_PUSH_CRED_REFRESH_*).
|
||||
// `run_turn` executes in the worker, so these must survive `env_clear()` to
|
||||
// reach the refresh-ahead loop in the ACP handler.
|
||||
EnvVars::FABRO_PUSH_CRED_REFRESH_AHEAD,
|
||||
EnvVars::FABRO_PUSH_CRED_REFRESH_INTERVAL_SECONDS,
|
||||
EnvVars::TERM,
|
||||
EnvVars::NO_COLOR,
|
||||
EnvVars::CLICOLOR,
|
||||
|
|
@ -113,6 +118,11 @@ mod tests {
|
|||
"FABRO_STORAGE_ROOT".to_string(),
|
||||
"/tmp/fabro-storage".to_string(),
|
||||
),
|
||||
("FABRO_PUSH_CRED_REFRESH_AHEAD".to_string(), "0".to_string()),
|
||||
(
|
||||
"FABRO_PUSH_CRED_REFRESH_INTERVAL_SECONDS".to_string(),
|
||||
"1800".to_string(),
|
||||
),
|
||||
("TERM".to_string(), "xterm-256color".to_string()),
|
||||
("NO_COLOR".to_string(), "1".to_string()),
|
||||
("CLICOLOR".to_string(), "0".to_string()),
|
||||
|
|
@ -147,6 +157,20 @@ mod tests {
|
|||
assert_eq!(actual.get("PATH").map(String::as_str), Some("/bin"));
|
||||
assert_eq!(actual.get("HOME").map(String::as_str), Some("/tmp/home"));
|
||||
assert_eq!(actual.get("FABRO_LOG").map(String::as_str), Some("debug"));
|
||||
// Push-credential refresh-ahead tunables must survive env_clear() into
|
||||
// the worker so run_turn's refresh-ahead loop can read them.
|
||||
assert_eq!(
|
||||
actual
|
||||
.get("FABRO_PUSH_CRED_REFRESH_AHEAD")
|
||||
.map(String::as_str),
|
||||
Some("0")
|
||||
);
|
||||
assert_eq!(
|
||||
actual
|
||||
.get("FABRO_PUSH_CRED_REFRESH_INTERVAL_SECONDS")
|
||||
.map(String::as_str),
|
||||
Some("1800")
|
||||
);
|
||||
assert_eq!(
|
||||
actual.get("TERM").map(String::as_str),
|
||||
Some("xterm-256color")
|
||||
|
|
|
|||
|
|
@ -22,6 +22,9 @@ impl EnvVars {
|
|||
pub const FABRO_LOG: &'static str = "FABRO_LOG";
|
||||
pub const FABRO_LOG_DESTINATION: &'static str = "FABRO_LOG_DESTINATION";
|
||||
pub const FABRO_NO_UPGRADE_CHECK: &'static str = "FABRO_NO_UPGRADE_CHECK";
|
||||
pub const FABRO_PUSH_CRED_REFRESH_AHEAD: &'static str = "FABRO_PUSH_CRED_REFRESH_AHEAD";
|
||||
pub const FABRO_PUSH_CRED_REFRESH_INTERVAL_SECONDS: &'static str =
|
||||
"FABRO_PUSH_CRED_REFRESH_INTERVAL_SECONDS";
|
||||
pub const FABRO_QUIET: &'static str = "FABRO_QUIET";
|
||||
pub const FABRO_SERVER: &'static str = "FABRO_SERVER";
|
||||
pub const FABRO_SERVER_MAX_CONCURRENT_RUNS: &'static str = "FABRO_SERVER_MAX_CONCURRENT_RUNS";
|
||||
|
|
@ -167,6 +170,8 @@ mod tests {
|
|||
EnvVars::FABRO_LOG,
|
||||
EnvVars::FABRO_LOG_DESTINATION,
|
||||
EnvVars::FABRO_NO_UPGRADE_CHECK,
|
||||
EnvVars::FABRO_PUSH_CRED_REFRESH_AHEAD,
|
||||
EnvVars::FABRO_PUSH_CRED_REFRESH_INTERVAL_SECONDS,
|
||||
EnvVars::FABRO_QUIET,
|
||||
EnvVars::FABRO_SERVER,
|
||||
EnvVars::FABRO_SERVER_MAX_CONCURRENT_RUNS,
|
||||
|
|
|
|||
|
|
@ -1,19 +1,26 @@
|
|||
//! Workflow adapter for ACP-backed LLM stages.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::env;
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::Duration;
|
||||
|
||||
use async_trait::async_trait;
|
||||
use fabro_acp::{
|
||||
AcpCommandError, AcpControlHandle, AcpError, AcpLiveControl, AcpProcessSpec, AcpRunRequest,
|
||||
render_stop_reason,
|
||||
};
|
||||
use fabro_agent::{AgentEvent, Sandbox, StaticEnvProvider, SteeringItem, ToolEnvProvider};
|
||||
use fabro_agent::{
|
||||
AgentEvent, RefreshOutcome, Sandbox, StaticEnvProvider, SteeringItem, ToolEnvProvider,
|
||||
};
|
||||
use fabro_graphviz::graph::Node;
|
||||
use fabro_static::EnvVars;
|
||||
use fabro_types::{
|
||||
AgentBackend, Principal, SessionCapability, StageId, StageTiming, SteeringMessage,
|
||||
};
|
||||
use fabro_util::time::elapsed_ms;
|
||||
use tokio::task::JoinHandle;
|
||||
use tokio::time::{sleep, timeout};
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
use super::super::agent::{CodergenBackend, CodergenResult, CodergenRunRequest, OneShotRequest};
|
||||
|
|
@ -24,6 +31,129 @@ use crate::event::{Emitter, Event, RunNoticeCode, RunNoticeLevel, StageScope};
|
|||
use crate::handler::NodeTimeoutPolicy;
|
||||
use crate::steering_hub::{ActiveControlHandle, SteeringHub};
|
||||
|
||||
/// Default refresh-ahead interval — comfortably under the ~60-min GitHub App
|
||||
/// installation-token TTL.
|
||||
const REFRESH_INTERVAL_DEFAULT: Duration = Duration::from_mins(45);
|
||||
/// Upper bound on a single push-credential refresh (token mint + `git remote
|
||||
/// set-url` exec). The turn-entry refresh runs before the ACP process spawns
|
||||
/// and the ACP node uses `NodeTimeoutPolicy::HandlerManaged`, so without this
|
||||
/// bound a stalled GitHub API call would hang node entry indefinitely.
|
||||
const REFRESH_MINT_TIMEOUT: Duration = Duration::from_secs(30);
|
||||
|
||||
/// Aborts the wrapped task when dropped, bounding the refresh-ahead loop to the
|
||||
/// lifetime of a single ACP turn.
|
||||
struct AbortOnDrop(JoinHandle<()>);
|
||||
impl Drop for AbortOnDrop {
|
||||
fn drop(&mut self) {
|
||||
self.0.abort();
|
||||
}
|
||||
}
|
||||
|
||||
/// Process-env lookup facade for the `FABRO_PUSH_CRED_REFRESH_*` tunables,
|
||||
/// isolated so the single disallowed-methods exception is documented in one
|
||||
/// place. Variable names come from [`EnvVars`].
|
||||
#[expect(
|
||||
clippy::disallowed_methods,
|
||||
reason = "Documented process-env facade for the FABRO_PUSH_CRED_REFRESH_* tunables; names come from fabro_static::EnvVars."
|
||||
)]
|
||||
fn refresh_env(name: &str) -> Option<String> {
|
||||
env::var(name).ok()
|
||||
}
|
||||
|
||||
/// Whether the push-credential refresh feature is enabled. Default ON; disabled
|
||||
/// by a falsy value (empty / `0` / `false` / `off` / `no`, case-insensitive),
|
||||
/// matching the repo's env-flag convention.
|
||||
fn parse_refresh_enabled(raw: Option<&str>) -> bool {
|
||||
!matches!(
|
||||
raw.map(|v| v.trim().to_ascii_lowercase()).as_deref(),
|
||||
Some("" | "0" | "false" | "off" | "no")
|
||||
)
|
||||
}
|
||||
|
||||
/// Parse the refresh-ahead loop interval. `None` disables the loop (explicit
|
||||
/// `0`, mirroring the codebase's `set_autostop_interval` "0 to disable"
|
||||
/// convention). Unset/empty or an unparsable value falls back to the default.
|
||||
fn parse_refresh_interval(raw: Option<&str>) -> Option<Duration> {
|
||||
match raw.map(str::trim) {
|
||||
None | Some("") => Some(REFRESH_INTERVAL_DEFAULT),
|
||||
Some(s) => match s.parse::<u64>() {
|
||||
Ok(0) => None,
|
||||
Ok(secs) => Some(Duration::from_secs(secs)),
|
||||
Err(_) => {
|
||||
tracing::warn!(
|
||||
value = %s,
|
||||
"invalid FABRO_PUSH_CRED_REFRESH_INTERVAL_SECONDS; using default"
|
||||
);
|
||||
Some(REFRESH_INTERVAL_DEFAULT)
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
fn push_cred_refresh_enabled() -> bool {
|
||||
parse_refresh_enabled(refresh_env(EnvVars::FABRO_PUSH_CRED_REFRESH_AHEAD).as_deref())
|
||||
}
|
||||
|
||||
fn push_cred_refresh_interval() -> Option<Duration> {
|
||||
parse_refresh_interval(
|
||||
refresh_env(EnvVars::FABRO_PUSH_CRED_REFRESH_INTERVAL_SECONDS).as_deref(),
|
||||
)
|
||||
}
|
||||
|
||||
/// Background loop that re-mints the sandbox's push credentials every
|
||||
/// `interval` for the duration of one ACP turn, so a single turn that outlives
|
||||
/// the installation-token TTL still pushes with a fresh token. Bounded by
|
||||
/// `cancel` (the drop-guard cancels it at turn end). A failed or timed-out tick
|
||||
/// retries after a shorter delay so a transient error does not leave a
|
||||
/// longer-than-interval window with an expired token.
|
||||
async fn refresh_ahead_loop(
|
||||
sandbox: Arc<dyn Sandbox>,
|
||||
cancel: CancellationToken,
|
||||
interval: Duration,
|
||||
) {
|
||||
let retry_delay = interval.min(Duration::from_mins(1));
|
||||
let mut delay = interval;
|
||||
loop {
|
||||
tokio::select! {
|
||||
() = cancel.cancelled() => break,
|
||||
() = sleep(delay) => {
|
||||
match timeout(REFRESH_MINT_TIMEOUT, sandbox.refresh_push_credentials())
|
||||
.await
|
||||
{
|
||||
Ok(Ok(RefreshOutcome::Refreshed)) => {
|
||||
tracing::info!(
|
||||
interval_secs = interval.as_secs(),
|
||||
"refresh-ahead re-minted push credentials mid-turn"
|
||||
);
|
||||
delay = interval;
|
||||
}
|
||||
Ok(Ok(RefreshOutcome::Skipped)) => {
|
||||
tracing::debug!(
|
||||
interval_secs = interval.as_secs(),
|
||||
"refresh-ahead tick: no managed push credentials to refresh"
|
||||
);
|
||||
delay = interval;
|
||||
}
|
||||
Ok(Err(e)) => {
|
||||
tracing::warn!(
|
||||
error = %fabro_sandbox::display_for_log(&e),
|
||||
"refresh-ahead mid-turn refresh failed; retrying sooner"
|
||||
);
|
||||
delay = retry_delay;
|
||||
}
|
||||
Err(_elapsed) => {
|
||||
tracing::warn!(
|
||||
timeout_secs = REFRESH_MINT_TIMEOUT.as_secs(),
|
||||
"refresh-ahead mid-turn refresh timed out; retrying sooner"
|
||||
);
|
||||
delay = retry_delay;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub struct AgentAcpBackend {
|
||||
tool_env: Option<Arc<dyn ToolEnvProvider>>,
|
||||
github_token_refresh_managed: bool,
|
||||
|
|
@ -137,6 +267,67 @@ impl AgentAcpBackend {
|
|||
}) as Arc<dyn Fn(String, Option<Principal>) + Send + Sync>
|
||||
});
|
||||
|
||||
// Keep the sandbox's push credentials fresh for the duration of this ACP
|
||||
// turn so the agent's own `git push` uses a live token instead of the one
|
||||
// baked into the clone at run start.
|
||||
//
|
||||
// Part 2 (turn-entry): re-mint + rewrite the origin URL before the ACP
|
||||
// process spawns, covering a push early in the turn. Non-fatal and
|
||||
// timeout-bounded — a stalled mint must neither fail nor hang node entry.
|
||||
// Part 3 (loop): a background task re-mints every ~45 min so a single turn
|
||||
// that itself outlives the ~60-min installation-token TTL still pushes
|
||||
// with a fresh token; a normal sub-interval turn never ticks (the
|
||||
// drop-guard aborts the task at turn end before the first tick).
|
||||
//
|
||||
// FABRO_PUSH_CRED_REFRESH_AHEAD=0 (or false/off/no/empty, case-
|
||||
// insensitive) disables the WHOLE feature — turn-entry re-mint AND loop —
|
||||
// so an operator who manages `origin` themselves can opt out of all
|
||||
// fabro-side origin rewriting. FABRO_PUSH_CRED_REFRESH_INTERVAL_SECONDS
|
||||
// overrides the loop interval; 0 disables just the loop.
|
||||
//
|
||||
// Known limitations tracked as follow-ups (not addressed here): (a)
|
||||
// resumed/parked runs reconnect the sandbox with no GitHub App creds, so
|
||||
// refresh no-ops until those creds are threaded through the reconnect
|
||||
// path; (b) the turn-entry re-mint has no freshness check, so it mints
|
||||
// once per node entry even when the current token is still fresh; (c) the
|
||||
// background `git remote set-url` can contend with the agent's own git on
|
||||
// `.git/config.lock`; (d) parallel ACP branches each run their own loop;
|
||||
// (e) this refresh lives in the ACP handler only, though the stale-origin
|
||||
// problem is stage-type-agnostic (native/command stages that push are not
|
||||
// covered); (f) refresh failures are logged via tracing but not surfaced
|
||||
// as a RunNotice event on the run stream.
|
||||
let refresh_enabled = push_cred_refresh_enabled();
|
||||
if refresh_enabled {
|
||||
match timeout(REFRESH_MINT_TIMEOUT, sandbox.refresh_push_credentials()).await {
|
||||
Ok(Ok(RefreshOutcome::Refreshed)) => {
|
||||
tracing::debug!("refreshed sandbox push credentials at ACP turn entry");
|
||||
}
|
||||
Ok(Ok(RefreshOutcome::Skipped)) => {}
|
||||
Ok(Err(e)) => {
|
||||
tracing::warn!(
|
||||
error = %fabro_sandbox::display_for_log(&e),
|
||||
"node-entry push-credential refresh failed (non-fatal)"
|
||||
);
|
||||
}
|
||||
Err(_elapsed) => {
|
||||
tracing::warn!(
|
||||
timeout_secs = REFRESH_MINT_TIMEOUT.as_secs(),
|
||||
"node-entry push-credential refresh timed out (non-fatal)"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
let _refresh_ahead_guard: Option<AbortOnDrop> = refresh_enabled
|
||||
.then(push_cred_refresh_interval)
|
||||
.flatten()
|
||||
.map(|interval| {
|
||||
AbortOnDrop(tokio::spawn(refresh_ahead_loop(
|
||||
Arc::clone(sandbox),
|
||||
cancel_token.child_token(),
|
||||
interval,
|
||||
)))
|
||||
});
|
||||
|
||||
let files_before = changed_files::detect_changed_files(sandbox).await;
|
||||
let launch_start = std::time::Instant::now();
|
||||
let result = match fabro_acp::run_acp_turn(AcpRunRequest {
|
||||
|
|
@ -416,21 +607,81 @@ mod tests {
|
|||
use std::collections::HashMap;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::Duration;
|
||||
|
||||
use fabro_acp::test_support::fake_acp_agent_script;
|
||||
use fabro_acp::{AcpError, AcpProcessExit};
|
||||
use fabro_agent::{LocalSandbox, Sandbox, shell_quote};
|
||||
use fabro_agent::{LocalSandbox, RefreshOutcome, Sandbox, shell_quote};
|
||||
use fabro_graphviz::graph::{AttrValue, Node};
|
||||
use fabro_sandbox::test_support::MockSandbox;
|
||||
use fabro_types::{CommandTermination, EventBody, ExecOutputTail};
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
use super::{AgentAcpBackend, acp_error_to_workflow};
|
||||
use super::{
|
||||
AgentAcpBackend, acp_error_to_workflow, parse_refresh_enabled, parse_refresh_interval,
|
||||
};
|
||||
use crate::context::Context;
|
||||
use crate::event::Emitter;
|
||||
use crate::handler::agent::{CodergenBackend, CodergenResult, CodergenRunRequest};
|
||||
use crate::steering_hub::SteeringHub;
|
||||
|
||||
#[test]
|
||||
fn refresh_enabled_defaults_on_and_honors_falsy_values() {
|
||||
// Default ON when unset.
|
||||
assert!(parse_refresh_enabled(None));
|
||||
// Truthy / non-falsy values stay enabled.
|
||||
for v in ["1", "true", "on", "yes", "anything"] {
|
||||
assert!(parse_refresh_enabled(Some(v)), "{v} should be enabled");
|
||||
}
|
||||
// Falsy values disable — case-insensitive, and empty/whitespace counts.
|
||||
for v in [
|
||||
"0", "false", "off", "no", "FALSE", "Off", "No", "OFF", "", " ",
|
||||
] {
|
||||
assert!(!parse_refresh_enabled(Some(v)), "{v} should be disabled");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn refresh_interval_parses_default_disable_and_override() {
|
||||
// Unset or empty → default.
|
||||
assert_eq!(parse_refresh_interval(None), Some(Duration::from_mins(45)));
|
||||
assert_eq!(
|
||||
parse_refresh_interval(Some(" ")),
|
||||
Some(Duration::from_mins(45))
|
||||
);
|
||||
// Explicit 0 disables the loop.
|
||||
assert_eq!(parse_refresh_interval(Some("0")), None);
|
||||
// A positive value overrides.
|
||||
assert_eq!(
|
||||
parse_refresh_interval(Some("1800")),
|
||||
Some(Duration::from_mins(30))
|
||||
);
|
||||
assert_eq!(
|
||||
parse_refresh_interval(Some(" 900 ")),
|
||||
Some(Duration::from_mins(15))
|
||||
);
|
||||
// Unparsable → default (never panics).
|
||||
for v in ["15m", "-1", "abc", "9999999999999999999999"] {
|
||||
assert_eq!(
|
||||
parse_refresh_interval(Some(v)),
|
||||
Some(Duration::from_mins(45)),
|
||||
"{v} should fall back to default"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn refresh_reports_skipped_without_managed_credentials() {
|
||||
// MockSandbox uses the trait default (no GitHub App creds), so refresh is
|
||||
// a no-op that must report Skipped — the signal the refresh-ahead loop
|
||||
// relies on to log at debug rather than falsely claim a re-mint.
|
||||
let sandbox = MockSandbox::linux();
|
||||
assert_eq!(
|
||||
sandbox.refresh_push_credentials().await.unwrap(),
|
||||
RefreshOutcome::Skipped
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn acp_backend_run_sends_prompt_and_returns_text() {
|
||||
let tempdir = tempfile::tempdir().unwrap();
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue