Make credential requests durable and store proxy authentication

host/src/cred/requests.rs

@@ -182,6 +182,7 @@ impl Cred {
         if let Some(expired) = expired_id {
             let _ = self.decided.send(expired);
         }
+        let _ = self.filed.send(id.clone());
         Ok(id)
     }
     pub fn request_list(&self) -> Result<Vec<Request>> {

host/src/cred/store.rs

@@ -16,6 +16,7 @@ pub(crate) struct Store {
 pub struct Cred {
     pub(crate) store: Mutex<Store>,
     pub(crate) decided: tokio::sync::broadcast::Sender<String>,
+    pub(crate) filed: tokio::sync::broadcast::Sender<String>,
 }
 impl Cred {
     pub fn open(host_dir: &Path) -> Result<Self> {
@@ -50,9 +51,11 @@ impl Cred {
         db.busy_timeout(std::time::Duration::from_secs(5))?;
         db.execute_batch(schema::DDL)?;
         let (decided, _) = tokio::sync::broadcast::channel(64);
+        let (filed, _) = tokio::sync::broadcast::channel(64);
         let cred = Self {
             store: Mutex::new(Store { db, master }),
             decided,
+            filed,
         };
         cred.expire_requests()?;
         Ok(cred)
@@ -60,6 +63,9 @@ impl Cred {
     pub(crate) fn access(&self) -> Result<MutexGuard<'_, Store>> {
         self.store.lock().map_err(|_| Error("storage"))
     }
+    pub fn subscribe_filed(&self) -> tokio::sync::broadcast::Receiver<String> {
+        self.filed.subscribe()
+    }
     pub fn subscribe(&self) -> tokio::sync::broadcast::Receiver<String> {
         self.decided.subscribe()
     }

host/src/harness/credentials.rs

@@ -0,0 +1,162 @@
+//! Credential request snapshots and host-only model proxy configuration.
+use crate::cred::{store::Cred, types::Subject};
+use deno_core::{OpState, op2};
+use deno_error::JsErrorBox;
+use serde_json::{Value, json};
+use std::{cell::RefCell, rc::Rc, sync::Arc};
+
+pub(super) struct Notifications {
+    filed: tokio::sync::Mutex<tokio::sync::broadcast::Receiver<String>>,
+}
+impl Notifications {
+    pub(super) fn new(cred: &Cred) -> Rc<Self> {
+        Rc::new(Self {
+            filed: tokio::sync::Mutex::new(cred.subscribe_filed()),
+        })
+    }
+}
+#[op2]
+pub(super) async fn op_credential_next(state: Rc<RefCell<OpState>>) {
+    let notifications = state.borrow().borrow::<Rc<Notifications>>().clone();
+    let mut filed = notifications.filed.lock().await;
+    let _ = tokio::time::timeout(std::time::Duration::from_secs(1), filed.recv()).await;
+}
+
+#[op2]
+#[serde]
+pub(super) fn op_credential_requests(state: &mut OpState) -> Result<serde_json::Value, JsErrorBox> {
+    let cred = state.borrow::<Arc<Cred>>();
+    let work = || -> anyhow::Result<Value> {
+        cred.expire_requests()?;
+        // TODO(T27): prune retained records once a day after the library adds Cred::prune.
+        let mut rows = Vec::new();
+        for request in cred.request_list()? {
+            let Subject::Conversation(conversation) = request.subject else {
+                continue;
+            };
+            let names = request
+                .payload
+                .get("items")
+                .and_then(Value::as_array)
+                .map(|items| {
+                    items
+                        .iter()
+                        .filter_map(Value::as_str)
+                        .map(|id| {
+                            cred.item_get(id)
+                                .map(|item| format!("`{}`", item.name))
+                                .unwrap_or_else(|_| "the credential".into())
+                        })
+                        .collect::<Vec<_>>()
+                        .join(", ")
+                })
+                .unwrap_or_else(|| "the credential change".into());
+            let message = match request.state.as_str() {
+                "approved" => format!("Your request for {names} was approved. Retry the command."),
+                "denied" => format!("Your request for {names} was denied."),
+                "expired" => format!(
+                    "Your request for {names} expired. File a new request if you still need it."
+                ),
+                _ => String::new(),
+            };
+            rows.push(json!({"id":request.id,"conversation":conversation,"state":request.state,"message":message}));
+        }
+        Ok(json!(rows))
+    };
+    work().map_err(|_| JsErrorBox::generic("Cannot read credential requests"))
+}
+
+pub(super) fn proxy(cred: &Cred) -> anyhow::Result<Option<Value>> {
+    use crate::cred::{
+        items::{Field, Item},
+        types::{Caller, FieldType},
+    };
+    let caller = Caller {
+        pid: 0,
+        program: "host".into(),
+        conversations: vec![],
+    };
+    let item = cred
+        .item_list()?
+        .into_iter()
+        .find(|item| item.name == "llm-proxy");
+    let item = match item {
+        Some(item) => item,
+        None => {
+            let (Ok(base), Ok(key)) = (
+                std::env::var("PI_PROXY_BASE_URL"),
+                std::env::var("PI_PROXY_API_KEY"),
+            ) else {
+                return Ok(None);
+            };
+            let mut store = cred.access()?;
+            let master = store.master.clone();
+            let tx = store
+                .db
+                .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
+            let mut item = crate::cred::items::insert_new(
+                &tx,
+                Item {
+                    id: String::new(),
+                    name: "llm-proxy".into(),
+                    version: 0,
+                    free_use: false,
+                    rules: vec![],
+                    validate_js: None,
+                    transform_js: "export default (ctx) => ctx.request".into(),
+                    fields: vec![
+                        Field {
+                            field_id: String::new(),
+                            name: "base_url".into(),
+                            public: true,
+                            value: Some(base),
+                            needs_value: false,
+                            field_type: FieldType::Text,
+                            key_type: None,
+                        },
+                        Field {
+                            field_id: String::new(),
+                            name: "api_key".into(),
+                            public: false,
+                            value: None,
+                            needs_value: true,
+                            field_type: FieldType::Text,
+                            key_type: None,
+                        },
+                    ],
+                },
+                &caller,
+                Some("host bootstrap"),
+            )?;
+            let field = &item
+                .fields
+                .iter()
+                .find(|f| f.name == "api_key")
+                .unwrap()
+                .field_id;
+            let sealed =
+                crate::cred::seal::seal(&master, &format!("{}|{field}", item.id), key.as_bytes())?;
+            tx.execute(
+                "INSERT INTO fields VALUES(?1,?2,?3)",
+                rusqlite::params![item.id, field, sealed],
+            )?;
+            item.fields
+                .iter_mut()
+                .find(|f| f.name == "api_key")
+                .unwrap()
+                .needs_value = false;
+            crate::cred::items::save(&tx, &item)?;
+            tx.commit()?;
+            item
+        }
+    };
+    Ok(Some(cred.credential(&item, true)?))
+}
+
+#[op2]
+#[serde]
+pub(super) fn op_model_proxy(state: &mut OpState) -> Result<serde_json::Value, JsErrorBox> {
+    proxy(state.borrow::<Arc<Cred>>())
+        .map(|v| v.unwrap_or(Value::Null))
+        .map_err(|_| JsErrorBox::generic("Cannot read llm-proxy credential"))
+}

host/src/harness/mod.rs

@@ -1,4 +1,5 @@
 //! Durable V8 harness, host-side SQLite storage, and world tool operations.
+mod credentials;
 mod environment;
 use crate::{exec::Exec, world::World};
 use deno_core::{JsRuntime, OpState, RuntimeOptions, op2};
@@ -78,6 +79,9 @@ deno_core::extension!(
     storage,
     ops = [
         op_sql,
+        credentials::op_credential_requests,
+        credentials::op_model_proxy,
+        credentials::op_credential_next,
         environment::op_file,
         environment::op_exec,
         environment::op_exec_next,
@@ -112,6 +116,18 @@ pub async fn run_with_exec(world: World, exec: Arc<dyn Exec>) -> anyhow::Result<
 /// Evaluate a bundled harness module with the same storage and tool operations as the console.
 /// The caller must use a current-thread Tokio runtime.
 pub async fn run_module(world: World, exec: Arc<dyn Exec>, source: &str) -> anyhow::Result<()> {
+    let cred = Arc::new(crate::cred::store::Cred::open(&world.host_dir)?);
+    run_module_with_credentials(world, exec, cred, source).await
+}
+
+/// Run a harness with the credentials handle shared by the host API.
+pub async fn run_module_with_credentials(
+    world: World,
+    exec: Arc<dyn Exec>,
+    cred: Arc<crate::cred::store::Cred>,
+    source: &str,
+) -> anyhow::Result<()> {
+    let proxy = credentials::proxy(&cred)?;
     let mut runtime = JsRuntime::new(RuntimeOptions {
         extensions: vec![
             deno_telemetry::deno_telemetry::init(),
@@ -162,11 +178,11 @@ pub async fn run_module(world: World, exec: Arc<dyn Exec>, source: &str) -> anyh
             deno_permissions::RuntimePermissionDescriptorParser::new(sys_traits::impls::RealSys),
         )));
     runtime.execute_script("bootstrap.js", include_str!("../../js/bootstrap.js"))?;
-    let env: serde_json::Map<String, Value> = ["PI_PROXY_BASE_URL", "PI_PROXY_API_KEY"]
-        .into_iter()
-        .filter_map(|key| std::env::var(key).ok().map(|v| (key.to_owned(), json!(v))))
-        .collect();
-    runtime.execute_script("env.js", format!("process.env = {};", Value::Object(env)))?;
+    runtime
+        .op_state()
+        .borrow_mut()
+        .put(credentials::Notifications::new(&cred));
+    runtime.op_state().borrow_mut().put(cred);
     runtime
         .op_state()
         .borrow_mut()
@@ -186,11 +202,26 @@ pub async fn run_module(world: World, exec: Arc<dyn Exec>, source: &str) -> anyh
         .load_main_es_module_from_code(&specifier, source.to_owned())
         .await?;
     let result = runtime.mod_evaluate(module);
-    runtime
-        .with_event_loop_promise(result, Default::default())
-        .await?;
-    runtime.run_event_loop(Default::default()).await?;
-    Ok(())
+    let result = async {
+        runtime
+            .with_event_loop_promise(result, Default::default())
+            .await?;
+        runtime.run_event_loop(Default::default()).await?;
+        Ok::<_, anyhow::Error>(())
+    }
+    .await;
+    result.map_err(|error| {
+        let mut text = error.to_string();
+        if let Some(key) = proxy
+            .as_ref()
+            .and_then(|v| v.get("api_key"))
+            .and_then(Value::as_str)
+            && !key.is_empty()
+        {
+            text = text.replace(key, "[redacted]");
+        }
+        anyhow::anyhow!(text)
+    })
 }
 
 /// Open a world and resume its durable root conversation.