use std::collections::BTreeSet;
use std::sync::Mutex;
use pagesift_api::types::{
InteractRequest, InteractResponse, InteractStopped, SESSION_REPLAYED, session_gone,
};
use pagesift_cdp::KeptPage;
use serde_json::Value;
use super::*;
mod act;
mod plan;
mod registry;
pub use registry::InteractLimits;
pub(crate) use registry::Sessions;
use registry::{OpenPage, PageState, Record};
tokio::task_local! {
pub(super) static KEEP: Arc<KeepSlot>;
}
#[derive(Default)]
pub(super) struct KeepSlot(Mutex<Option<(KeptPage, Arc<BrowserPool>)>>);
impl KeepSlot {
pub(super) fn put(&self, page: KeptPage, pool: Arc<BrowserPool>) {
*self.0.lock().unwrap_or_else(|e| e.into_inner()) = Some((page, pool));
}
fn take(&self) -> Option<(KeptPage, Arc<BrowserPool>)> {
self.0.lock().unwrap_or_else(|e| e.into_inner()).take()
}
}
pub(super) fn interact_capable(cfg: &PoolConfig) -> bool {
cfg.engine.unwrap_or(pagesift_cdp::EngineKind::Chrome) == pagesift_cdp::EngineKind::Chrome
}
pub(super) fn record_budget(chain: &[PoolConfig]) -> usize {
let mb = chain
.iter()
.map(|c| {
let f = c
.engine
.unwrap_or(pagesift_cdp::EngineKind::Chrome)
.footprint();
f.browser_mb + f.tab_mb
})
.max()
.unwrap_or(0) as usize;
(mb * 1024 * 1024).min(MAX_RECORD_BYTES)
}
const MAX_RECORD_BYTES: usize = 16 * 1024 * 1024;
fn redact(text: &str, secrets: &[(&str, &str)]) -> String {
let mut out = text.to_string();
for (name, value) in secrets {
if value.is_empty() {
continue;
}
let holder = format!("{{{{secret:{name}}}}}");
let encoded: String = url::form_urlencoded::byte_serialize(value.as_bytes()).collect();
let escaped = value
.replace('&', "&")
.replace('<', "<")
.replace('>', ">")
.replace('"', """);
for form in [value, &encoded.as_str(), &escaped.as_str()] {
if out.contains(*form) {
out = out.replace(*form, &holder);
}
}
}
out
}
fn redact_json(v: &mut Value, secrets: &[(&str, &str)]) {
match v {
Value::String(s) => *s = redact(s, secrets),
Value::Array(a) => a.iter_mut().for_each(|x| redact_json(x, secrets)),
Value::Object(m) => m.values_mut().for_each(|x| redact_json(x, secrets)),
_ => {}
}
}
fn has_secret(action: &Value) -> bool {
action
.get("text")
.and_then(Value::as_str)
.is_some_and(|t| t.contains("{{secret:"))
}
fn is_script(action: &Value) -> bool {
action.get("type").and_then(Value::as_str) == Some("executeJavascript")
}
fn secret_names(text: &str) -> impl Iterator<Item = &str> {
text.split("{{secret:")
.skip(1)
.filter_map(|rest| rest.split_once("}}").map(|(name, _)| name))
}
impl Scraper {
pub(super) fn interact_available(&self) -> bool {
self.browsers.iter().any(|(c, _)| interact_capable(c))
}
fn interact_unavailable(&self) -> ApiError {
ApiError::unsupported(
"interact needs Chrome: no Chrome or Chromium was found on this machine (run `pagesift doctor`)",
)
}
async fn interact_pool(&self) -> Result<Arc<BrowserPool>, ApiError> {
for i in 0..self.browsers.len() {
if interact_capable(&self.browsers[i].0)
&& let Some(pool) = self.browser(i).await
{
return Ok(pool);
}
}
Err(self.interact_unavailable())
}
async fn admit_page(&self) -> Result<registry::Reserved, ApiError> {
let pool = self.interact_pool().await?;
let cap = self.sessions.sweep(Some(pool.session_room()), 1).await;
self.sessions.reserve(cap).map_err(|open| {
ApiError::new(
ErrorCode::ConcurrencyLimitExceeded,
format!(
"cannot keep another browser session: {open} are open and in use, and this machine's memory and browser tabs admit {cap}"
),
)
})
}
async fn open_page(
&self,
url: &str,
opts: &ScrapeOptions,
mode: RobotsMode,
) -> Result<(Document, KeptPage, Arc<BrowserPool>), ApiError> {
let mut inner = opts.clone();
inner.engine = Some("browser".into());
inner.max_age = Some(0);
inner.store_in_cache = false;
let slot = Arc::new(KeepSlot::default());
let doc = KEEP
.scope(slot.clone(), self.scrape_with(url, &inner, mode))
.await?;
let (page, pool) = slot.take().ok_or_else(|| {
ApiError::unsupported(
"this page was answered without a browser (a rule chose the engine, or the content is not a web page), so there is no session to keep",
)
})?;
Ok((doc, page, pool))
}
fn opened(
&self,
page: KeptPage,
pool: Arc<BrowserPool>,
doc: &Document,
url: &str,
opts: &ScrapeOptions,
) -> Box<OpenPage> {
self.sessions.start_reaper();
let now = Instant::now();
Box::new(OpenPage {
page,
pool,
opened: now,
last_used: now,
url: doc.metadata.url.clone().unwrap_or_else(|| url.to_string()),
secrets: BTreeSet::new(),
scripted: opts.actions.iter().any(is_script),
})
}
pub async fn scrape_session(
&self,
url: &str,
opts: &ScrapeOptions,
keep: bool,
owner: Option<&str>,
) -> Result<Document, ApiError> {
if !self.interact_available() {
if keep {
return Err(self.interact_unavailable());
}
return self.scrape(url, opts).await;
}
let mode = self
.policy
.robots_mode(RequestKind::Scrape, opts.ignore_robots_txt)
.map_err(policy_error)?;
if !keep {
let mut doc = self.scrape_with(url, opts, mode).await?;
let id = self
.sessions
.insert(None, owner, url, opts, mode, PageState::NotOpened);
doc.metadata.scrape_id = Some(id);
return Ok(doc);
}
if opts.engine.as_deref() == Some("http") {
return Err(ApiError::invalid(
"keepSession keeps a browser page and cannot be combined with engine \"http\"",
));
}
let _reserved = self.admit_page().await?;
let (mut doc, page, pool) = self.open_page(url, opts, mode).await?;
let open = self.opened(page, pool, &doc, url, opts);
let id = self
.sessions
.insert(None, owner, url, opts, mode, PageState::Open(open));
doc.metadata.scrape_id = Some(id);
Ok(doc)
}
fn record_for(
&self,
id: &str,
req: &InteractRequest,
owner: Option<&str>,
) -> Result<Arc<Record>, ApiError> {
match (self.sessions.get(id, owner), req.url.as_deref()) {
(Ok(r), _) => Ok(r),
(Err(e), Some(url))
if matches!(e.code, ErrorCode::SessionGone | ErrorCode::NotFound)
&& pagesift_api::types::is_scrape_id(id) =>
{
let opts = ScrapeOptions::default();
let mode = self
.policy
.robots_mode(RequestKind::Scrape, opts.ignore_robots_txt)
.map_err(policy_error)?;
self.sessions
.insert(Some(id), owner, url, &opts, mode, PageState::NotOpened);
self.sessions.get(id, owner)
}
(Err(e), _) => Err(e),
}
}
pub async fn interact(
&self,
id: &str,
req: &InteractRequest,
owner: Option<&str>,
) -> Result<InteractResponse, ApiError> {
req.check()?;
if !self.interact_available() {
return Err(self.interact_unavailable());
}
if req.prompt.is_some() && self.llm.is_none() {
return Err(ApiError::unsupported(
"prompt needs an LLM: set [llm] in pagesift.toml, or send `actions`",
));
}
let actions = req.actions.clone().unwrap_or_default();
let record = self.record_for(id, req, owner)?;
let mut resp = InteractResponse::default();
let limit = req.time_limit();
let deadline = tokio::time::Instant::now() + limit;
let timed_out = |resp: &mut InteractResponse| {
resp.killed = true;
resp.fail(format!(
"the call did not finish within {} s; the session is still open",
limit.as_secs_f64()
));
};
let Ok(mut state) = tokio::time::timeout_at(deadline, record.state.lock()).await else {
timed_out(&mut resp);
return Ok(resp);
};
if let PageState::Gone(why) = &*state {
return Err(session_gone(format!(
"the browser session of scrape {id} has ended: {why}"
)));
}
if matches!(&*state, PageState::NotOpened) {
let opts = replay_options(&record.opts, limit);
let opening = async {
let reserved = self.admit_page().await?;
let opened = self.open_page(&record.url, &opts, record.mode).await?;
Ok::<_, ApiError>((reserved, opened))
};
let (_reserved, (doc, page, pool)) =
match tokio::time::timeout_at(deadline, opening).await {
Ok(r) => r?,
Err(_) => {
timed_out(&mut resp);
return Ok(resp);
}
};
*state = PageState::Open(self.opened(page, pool, &doc, &record.url, &opts));
self.sessions.mark_open(id, &record);
resp.warnings.push(format!(
"{SESSION_REPLAYED}: the scrape did not keep its page (keepSession), so {} was opened again; the page may differ from the scrape result",
record.url
));
}
let PageState::Open(open) = &mut *state else {
return Err(session_gone(format!(
"the browser session of scrape {id} has ended"
)));
};
let run = async {
match &req.prompt {
Some(prompt) => self.run_prompt(&record, open, prompt, &mut resp).await,
None => self.run_actions(&record, open, actions, &mut resp).await,
}
};
let outcome = tokio::time::timeout_at(deadline, run).await;
open.last_used = Instant::now();
let alive = open.page.is_alive();
let redacting: Vec<(String, String)> = open
.secrets
.iter()
.filter_map(|n| self.secrets.get(n).map(|v| (n.clone(), v.value.clone())))
.collect();
let outcome = outcome.map(|r| {
r.map_err(|mut e| {
let pairs: Vec<(&str, &str)> = redacting
.iter()
.map(|(n, v)| (n.as_str(), v.as_str()))
.collect();
e.message = redact(&e.message, &pairs);
e
})
});
match outcome {
Err(_) => timed_out(&mut resp),
Ok(Ok(())) => {}
Ok(Err(e)) => {
let closes = e.code == ErrorCode::RobotsDisallowed || !alive;
if closes {
let why = if alive {
"an action navigated to a page robots.txt disallows".to_string()
} else {
"the browser exited".to_string()
};
let old = std::mem::replace(&mut *state, PageState::Gone(why.clone()));
drop(state);
if let PageState::Open(o) = old {
o.page.close().await;
}
self.sessions.mark_closed(id);
return Err(if alive {
e
} else {
session_gone(format!(
"the browser session of scrape {id} has ended: {why}"
))
});
}
return Err(e);
}
}
drop(state);
finish_response(resp, &redacting)
}
pub async fn stop_interact(
&self,
id: &str,
owner: Option<&str>,
) -> Result<InteractStopped, ApiError> {
let not_found = |detail: &str| {
ApiError::new(
ErrorCode::NotFound,
format!("no browser session for scrape {id}: {detail}"),
)
};
let record = self.sessions.get(id, owner).map_err(|e| match e.code {
ErrorCode::SessionGone => ApiError::new(ErrorCode::NotFound, e.message),
_ => e,
})?;
let mut state = record.state.lock().await;
match &*state {
PageState::NotOpened => return Err(not_found("no session was started")),
PageState::Gone(why) => return Err(not_found(&format!("it has ended: {why}"))),
PageState::Open(_) => {}
}
let old = std::mem::replace(&mut *state, PageState::Gone("it was stopped".into()));
drop(state);
let PageState::Open(open) = old else {
return Err(not_found("no session was started"));
};
let held = open.opened.elapsed();
open.page.close().await;
self.sessions.mark_closed(id);
Ok(InteractStopped {
success: true,
session_duration_ms: held.as_millis() as u64,
credits_billed: 0,
})
}
async fn polite_permit(
&self,
record: &Record,
current: &str,
) -> Result<pagesift_policy::Permit, ApiError> {
let target = Url::parse(current).map_err(|e| ApiError::invalid(e.to_string()))?;
let delay = match record.mode {
RobotsMode::Respect => self
.policy
.check_robots(&target)
.await
.map_err(policy_error)?,
RobotsMode::Ignore => None,
};
let host = target.host_str().unwrap_or_default().to_string();
Ok(self.policy.limiter().acquire(&host, delay).await)
}
async fn run_actions(
&self,
record: &Record,
open: &mut OpenPage,
mut actions: Vec<Value>,
resp: &mut InteractResponse,
) -> Result<(), ApiError> {
pagesift_cdp::parse_actions(&Value::Array(actions.clone()))
.map_err(|e| ApiError::invalid(e.to_string()))?;
let filling = actions.iter().any(has_secret);
let scripting = actions.iter().any(is_script);
if scripting && (filling || !open.secrets.is_empty()) || filling && open.scripted {
return Err(ApiError::invalid(
"a session cannot both run executeJavascript and use {{secret:...}}",
));
}
open.scripted |= scripting;
for a in &actions {
if let Some(text) = a.get("text").and_then(Value::as_str) {
open.secrets.extend(secret_names(text).map(str::to_string));
}
}
let permit = self.polite_permit(record, &open.url.clone()).await?;
let started = Instant::now();
let mut out = pagesift_cdp::ActionOutputs::default();
let mut failed = None;
let mut rest = actions.as_mut_slice();
while !rest.is_empty() && failed.is_none() {
let n = if has_secret(&rest[0]) {
1
} else {
rest.iter().position(has_secret).unwrap_or(rest.len())
};
let (chunk, tail) = std::mem::take(&mut rest).split_at_mut(n);
rest = tail;
if has_secret(&chunk[0]) {
match open.page.capture().await {
Ok(page) => {
super::autoschema::fill_secrets(chunk, &self.secrets, &page.final_url)?
}
Err(e) => {
failed = Some(e);
break;
}
}
}
let parsed = pagesift_cdp::parse_actions(&Value::Array(chunk.to_vec()))
.map_err(|e| ApiError::invalid(e.to_string()))?;
match open.page.run_actions(&parsed).await {
Ok(o) => {
out.screenshots.extend(o.screenshots);
out.scrapes.extend(o.scrapes);
out.javascript_returns.extend(o.javascript_returns);
out.pdfs.extend(o.pdfs);
}
Err(e) => failed = Some(e),
}
}
match failed {
None => {
resp.result = out.javascript_returns.last().map(|j| j.value.to_string());
resp.actions = Some(out.to_firecrawl_json());
}
Some(e) => action_failure(e, resp)?,
}
drop(permit);
self.fill_page(record, open, resp, started).await
}
async fn fill_page(
&self,
record: &Record,
open: &mut OpenPage,
resp: &mut InteractResponse,
started: Instant,
) -> Result<(), ApiError> {
let mut page = match open.page.capture().await {
Ok(p) => p,
Err(e) => {
if open.page.is_alive() {
if resp.error.is_none() {
resp.fail(format!("could not read the page after the actions: {e}"));
}
return Ok(());
}
return Err(ApiError::new(render_error_code(&e), e.to_string()));
}
};
if page.final_url != open.url {
if record.mode == RobotsMode::Respect
&& let Ok(u) = Url::parse(&page.final_url)
{
self.policy.check_robots(&u).await.map_err(policy_error)?;
}
open.url = page.final_url.clone();
}
let secrets = self.named_secrets(&open.secrets);
if !secrets.is_empty() {
page.html = redact(&page.html, &secrets);
}
let opts = page_options(&record.opts);
let doc = self
.live_document(page, &record.url, &opts, Vec::new(), started)
.await?;
resp.data = Some(doc);
Ok(())
}
fn named_secrets<'a>(&'a self, names: &'a BTreeSet<String>) -> Vec<(&'a str, &'a str)> {
names
.iter()
.filter_map(|n| self.secrets.get(n).map(|v| (n.as_str(), v.value.as_str())))
.collect()
}
}
fn action_failure(e: pagesift_cdp::Error, resp: &mut InteractResponse) -> Result<(), ApiError> {
use pagesift_cdp::Error as E;
match e {
E::Action { .. } | E::Protocol { .. } => {
resp.fail(e.to_string());
Ok(())
}
E::InvalidOptions(m) => Err(ApiError::invalid(m)),
E::ExecuteJavascriptDisabled => Err(ApiError::unsupported(e.to_string())),
other => Err(ApiError::new(render_error_code(&other), other.to_string())),
}
}
fn replay_options(opts: &ScrapeOptions, limit: Duration) -> ScrapeOptions {
let mut o = opts.clone();
let ms = limit.as_millis() as u64;
o.timeout = Some(o.timeout.map_or(ms, |t| t.min(ms)));
o
}
fn page_options(opts: &ScrapeOptions) -> ScrapeOptions {
let mut o = opts.clone();
o.formats.retain(|f| {
matches!(
f.kind.as_str(),
"markdown" | "html" | "rawHtml" | "links" | "images"
)
});
if o.formats.is_empty() {
o.formats.push(pagesift_api::types::Format::new("markdown"));
}
o
}
fn finish_response(
resp: InteractResponse,
secrets: &[(String, String)],
) -> Result<InteractResponse, ApiError> {
if secrets.is_empty() {
return Ok(resp);
}
let pairs: Vec<(&str, &str)> = secrets
.iter()
.map(|(n, v)| (n.as_str(), v.as_str()))
.collect();
let mut v = serde_json::to_value(&resp)
.map_err(|e| ApiError::new(ErrorCode::Internal, e.to_string()))?;
redact_json(&mut v, &pairs);
serde_json::from_value(v).map_err(|e| ApiError::new(ErrorCode::Internal, e.to_string()))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn secrets_are_named_and_redacted() {
let names: Vec<&str> = secret_names("log in as {{secret:USER}} / {{secret:PW}}").collect();
assert_eq!(names, ["USER", "PW"]);
assert_eq!(secret_names("nothing here").count(), 0);
let pairs = [("PW", "hunter2"), ("EMPTY", "")];
assert_eq!(redact("value=hunter2;", &pairs), "value={{secret:PW}};");
let mut v = serde_json::json!({"a": ["hunter2"], "b": {"c": "x hunter2"}, "n": 1});
redact_json(&mut v, &pairs);
assert_eq!(
v,
serde_json::json!({"a": ["{{secret:PW}}"], "b": {"c": "x {{secret:PW}}"}, "n": 1})
);
}
#[test]
fn page_options_drop_llm_formats() {
let o: ScrapeOptions =
serde_json::from_value(serde_json::json!({"formats": ["summary", "links"]})).unwrap();
let kinds: Vec<_> = page_options(&o)
.formats
.iter()
.map(|f| f.kind.clone())
.collect();
assert_eq!(kinds, ["links"]);
let o: ScrapeOptions =
serde_json::from_value(serde_json::json!({"formats": ["summary"]})).unwrap();
assert!(page_options(&o).wants("markdown"));
}
}