feat: stop a session when its grants change, and keep its policy current

PLAN.md

@@ -70,7 +70,8 @@ Landed so far:
 | 1 | #71 | Nothing |
 | 2 | #73, and attempt keys for a claim and a completion in #77 | An attempt key on the other operations; an uncertain external effect blocking reassignment; waiting on an outside event as a state; the record's remaining fields (`parent`, `scores`, `implements`, `reversibility`, `artifacts_touched`, `policy_context`) |
 | 3 | #74 | A suite that gates a version of a definition; the data scope enforced; a definition required by default, once the system's own filings name one; tools offered a few at boot and the rest by search |
-| 6 | #78 | `check_policy` selecting rules by the action; the unused-rule check over one review period; the approver of an exception checked against the layer breached; a new version of a rule keeping its place under the ceiling; a decision that survives an enforced module that fails; the author and the date of an imported rule not trusted; a session that reloads the policy when a rule changes; a verified rule that stays in force while its next version waits for review; a denial that names no policy counted for the module that made it, in the rates and the lint |
+| 6 | #78 | `check_policy` selecting rules by the action; the unused-rule check over one review period; the approver of an exception checked against the layer breached; a new version of a rule keeping its place under the ceiling; a decision that survives an enforced module that fails; the author and the date of an imported rule not trusted; a verified rule that stays in force while its next version waits for review; a denial that names no policy counted for the module that made it, in the rates and the lint |
+| 10 | #79 | A stop for only the sessions that a change to the grants affects; rendered files removed when a session stops; the harness of a stopped session stopped; an expiry on a grant; an expiry that reaches a read filter from a root policy, and the advisory rules |
 
 What stays unbuilt after these, and stays on README's list: federation across domains, authentication of people, logprob-based drift, and accelerator time, which only a gateway can count. So do the items on that list that need another machine or an outside service more than they need code: the limits on re-enactment, harness settings and decomposition templates in the platform domain, bytes stored and resource limits on the sandbox, the authoring of agent definitions, the search tool against its live provider, a harness under Seatbelt on a Mac, and the Gofannon bridge against a live deployment, whose agents make model calls that the egress does not meter.
 

README.md

@@ -142,8 +142,12 @@ two weeks. It is also the reason that none of it is the production shape.
 - **Curation and audit, in part.** No curation task demotes an entry, checks an owner or
   a cross-link, or curates policy. No test on planted faults qualifies the auditor's
   model. The auditor does not read handoffs.
-- **Access freshness.** A session keeps its read filter until it ends. A change to the
-  grants or to a policy does not stop a session.
+- **Access freshness, in part.** Any change to the grants file stops each open session,
+  not only a session that the change affects. The files that a session rendered before
+  the change stay on disk. A harness that a stopped session started continues to run,
+  and each of its memory calls and model calls is refused. A read that waits does not
+  check again before it gives its answer. Some commands of the owner do not do the
+  check. A grant has no expiry of its own.
 - **The improvement loop.** The only gate on a platform artifact is that a person
   verifies it. There is no performance gate, no safety gate, and no check that a change
   adds no permission.

crates/egress/src/lib.rs

@@ -235,6 +235,8 @@ struct Inner {
     /// How long a tunnel has for the TLS hellos the egress reads, [`hello::HELLO_WAIT`]
     /// outside a test.
     hello_wait: std::time::Duration,
+    /// How often an open tunnel asks whether its session is still valid.
+    stop_check: std::time::Duration,
 }
 
 /// How the egress resolves a tunnel's host and dials it: the system's resolver and TCP,
@@ -377,6 +379,7 @@ impl Egress {
                 telemetry,
                 network: Arc::new(SystemNetwork),
                 hello_wait: hello::HELLO_WAIT,
+                stop_check: std::time::Duration::from_secs(1),
             }),
         })
     }
@@ -399,6 +402,10 @@ impl Egress {
         body: Bytes,
     ) -> Result<Relayed, Error> {
         let inner = &self.inner;
+        inner
+            .memory
+            .fresh()
+            .map_err(|e| Error::Refused(e.to_string()))?;
         let dest = inner
             .tools
             .get(tool)
@@ -834,13 +841,23 @@ fn over_budget(inner: &Inner, destination: &str) -> Option<String> {
         .telemetry
         .spent(destination, None, Some(day_ago))
         .unwrap_or(0);
+    // a check that cannot be made is a refusal: a session that has stopped, or a
+    // policy that does not evaluate, sends nothing
     match inner.memory.check_spend(destination, session, day) {
         Ok(d) if !d.allow => Some(d.reason),
-        _ => None,
+        Ok(_) => None,
+        Err(e) => Some(e.to_string()),
     }
 }
 
 fn admit(inner: &Inner, destination: &str) -> Result<Destination, Box<Response>> {
+    if let Err(e) = inner.memory.fresh() {
+        return Err(Box::new(error_json(
+            StatusCode::FORBIDDEN,
+            "permission_error",
+            &e.to_string(),
+        )));
+    }
     let Some(dest) = inner.destinations.get(destination) else {
         return Err(Box::new(error_json(
             StatusCode::NOT_FOUND,
@@ -1033,6 +1050,9 @@ async fn tunnel_for(
             message,
         ))
     };
+    if let Err(e) = inner.memory.fresh() {
+        return Err(refuse(&e.to_string()));
+    }
     let Some((host, port)) = host_and_port(authority) else {
         return Err(Box::new(error_json(
             StatusCode::BAD_REQUEST,
@@ -1110,7 +1130,24 @@ async fn connect_or_not_found(State(egress): State<Egress>, mut req: Request) ->
             }
             Err(Stop::Ended) => return,
         };
-        let _ = tokio::io::copy_bidirectional(&mut inbound, &mut outbound).await;
+        // an open tunnel is the session's still: it carries bytes only while the session
+        // is valid, and is closed when the session stops
+        let stopped = async {
+            loop {
+                tokio::time::sleep(inner.stop_check).await;
+                if inner.memory.fresh().is_err() {
+                    break;
+                }
+            }
+        };
+        tokio::select! {
+            _ = tokio::io::copy_bidirectional(&mut inbound, &mut outbound) => {}
+            () = stopped => {
+                eprintln!("egress: tunnel to {host}: the session stopped; closed");
+                let _ = inbound.shutdown().await;
+                let _ = outbound.shutdown().await;
+            }
+        }
         let _ = inner.telemetry.record(
             started,
             &session,
@@ -1883,6 +1920,56 @@ mod tests {
         ));
     }
 
+    /// A session whose grants changed has stopped: no model call of it leaves, by the
+    /// library or by the router.
+    #[tokio::test]
+    async fn a_stopped_session_sends_no_model_call() {
+        let fake = Router::new().route(
+            "/v1/chat/completions",
+            any(|| async {
+                (
+                    [(header::CONTENT_TYPE, "application/json")],
+                    r#"{"model":"m","choices":[{"message":{"content":"ok"}}],"usage":{"prompt_tokens":4,"completion_tokens":2}}"#,
+                )
+            }),
+        );
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        tokio::spawn(async move { axum::serve(listener, fake).await.unwrap() });
+        let dir = tempfile::tempdir().unwrap();
+        let root = dir.path().join("m");
+        Memory::init(&root, "human:test:me", "dom-me").unwrap();
+        let m = Arc::new(Memory::open(&root).unwrap().with_author("agent:a"));
+        let mut d = HashMap::new();
+        d.insert(
+            "openai".to_string(),
+            Destination::forward(format!("http://{addr}"), Auth::PassThrough),
+        );
+        let e = Egress::new(m, d, &dir.path().join("t.db")).unwrap();
+        let ask = serde_json::json!({ "messages": [{ "role": "user", "content": "hi" }] });
+        e.call("openai", "v1/chat/completions", &ask).await.unwrap();
+        let grants = root.join("grants.toml");
+        let text = std::fs::read_to_string(&grants).unwrap();
+        std::fs::write(&grants, format!("{text}\n# the owner changed the grants\n")).unwrap();
+        assert!(e.call("openai", "v1/chat/completions", &ask).await.is_err());
+        let r = e
+            .router()
+            .oneshot(
+                Request::builder()
+                    .uri("/openai/v1/chat/completions")
+                    .method("POST")
+                    .header("content-type", "application/json")
+                    .body(Body::from(ask.to_string()))
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+        assert_eq!(r.status(), StatusCode::FORBIDDEN);
+        let body = to_bytes(r.into_body(), 1 << 20).await.unwrap();
+        let said = String::from_utf8_lossy(&body);
+        assert!(said.contains("stopped"), "{said}");
+    }
+
     #[tokio::test]
     async fn a_model_call_over_the_spend_limit_does_not_leave() {
         let fake = Router::new().route(
@@ -3555,6 +3642,33 @@ mod tests {
         (answer, c)
     }
 
+    /// A tunnel that is open when its session stops is closed: a harness that keeps a
+    /// connection alive sends nothing more through it.
+    #[tokio::test]
+    async fn an_open_tunnel_is_closed_when_its_session_stops() {
+        use tokio::io::AsyncWriteExt;
+        let (addr, _) = echo_upstream().await;
+        let entry = format!("127.0.0.1:{}", addr.port());
+        let (d, mut e) = fixture(Some(("github", Destination::tunnel([entry])))).await;
+        Arc::get_mut(&mut e.inner).unwrap().stop_check = std::time::Duration::from_millis(50);
+        let egress = serve(&e).await;
+        let (head, mut c) = connect(egress, &addr.to_string()).await;
+        assert!(head.starts_with("HTTP/1.1 200"), "{head}");
+        echoes(&mut c, "127.0.0.1").await;
+        let grants = d.path().join("m").join(dm_server::GRANTS_FILE);
+        let text = std::fs::read_to_string(&grants).unwrap();
+        std::fs::write(&grants, format!("{text}\n# a change\n")).unwrap();
+        let (rest, closed) = until_closed(&mut c).await;
+        assert!(closed, "the tunnel stayed open after the session stopped");
+        assert_eq!(rest, b"");
+        // and nothing written now comes back
+        let _ = c.write_all(b"sent after the session stopped").await;
+        assert_eq!(until_closed(&mut c).await.0, b"");
+        // nor does a new tunnel open
+        let (head, _) = connect(egress, &addr.to_string()).await;
+        assert!(head.starts_with("HTTP/1.1 403"), "{head}");
+    }
+
     #[tokio::test]
     async fn a_connect_tunnel_opens_only_to_allowed_hosts_within_the_ceiling() {
         let (addr, _) = echo_upstream().await;

crates/index/src/lib.rs

@@ -277,6 +277,22 @@ impl Index {
         Index::init(conn)
     }
 
+    /// An index that exists, opened to read: nothing is created or changed, and a
+    /// writer elsewhere is waited for only briefly. For a reader that may be called
+    /// while another handle of the same process is at work.
+    ///
+    /// # Errors
+    ///
+    /// When the file is not there or cannot be opened.
+    pub fn open_reader(path: &Path) -> Result<Index, Error> {
+        let conn = Connection::open_with_flags(
+            path,
+            rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX,
+        )?;
+        conn.busy_timeout(std::time::Duration::from_millis(250))?;
+        Ok(Index { conn })
+    }
+
     /// An index that lives only in memory.
     ///
     /// # Errors