From fdcfa063c7da6de1b0f3f21a677619cb2633bada Mon Sep 17 00:00:00 2001
From: Meta Vens <rwvens@hotmail.com>
Date: Fri, 30 Oct 2026 00:11:53 +0100
Subject: [PATCH] connector-many-callers: one connector serves every Mind as
 its own caller; session handles are an HMAC under a connector secret over
 caller, session, instance, instructions, model and transcript prefix; config
 add-caller/add-lease-caller; one owner for distinct admissions (squash of
 eureka-body/connector-many-callers d1f25b5..a96b276; Soul s1-s2)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 Cargo.lock                                    |   1 +
 README.md                                     |   5 +-
 crates/claude-connector/Cargo.toml            |   1 +
 crates/claude-connector/src/backend.rs        |  83 ++-
 crates/claude-connector/src/backend/tests.rs  | 296 ++++++++++-
 crates/claude-connector/src/config.rs         |   3 +
 crates/claude-connector/src/daemon.rs         |  92 ++--
 crates/claude-connector/src/handle.rs         |  69 ++-
 crates/claude-connector/src/lib.rs            |   2 +-
 crates/claude-connector/src/main.rs           | 264 ++++++++--
 crates/claude-connector/src/quota.rs          |   2 +-
 .../claude-connector/tests/config_document.rs | 474 +++++++++++++++++-
 crates/model-connector/src/serve.rs           |  23 +-
 crates/model-connector/src/service.rs         |  63 ++-
 14 files changed, 1228 insertions(+), 150 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 524654e..5291d19 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -140,6 +140,7 @@ version = "0.1.0"
 dependencies = [
  "anyhow",
  "cultcache-rs",
+ "hmac",
  "libc",
  "model-connector",
  "rmp-serde",
diff --git a/README.md b/README.md
index 3a12abd..c2e23c7 100644
--- a/README.md
+++ b/README.md
@@ -42,8 +42,9 @@ Crate `crates/claude-connector` (library only; the binary that wires it to `serv
   servers, no settings sources and the system prompt replaced. Tool-less turns only: a request
   with tools or a final tool result is refused before any spawn. A request's output schema goes
   to the CLI as `--json-schema` and the result's structured output is the one assistant text.
-- Reasoning state is a session handle (`ClaudeCodeHandle`); a handle that cannot be proven to
-  match the request re-seeds from the caller's transcript, and every resume forks.
+- Reasoning state is a session handle (`ClaudeCodeHandle`) authenticated with a connector secret
+  and bound to the caller it was minted for; a handle that cannot be proven to match the request
+  and its caller re-seeds from the caller's transcript, and every resume forks.
 - `ProcessPort` is the one OS seam (`StdProcessPort`, unix, kills the child's whole process
   group); `SessionStore` and `StderrSink` are the other ports. `ClaudeCodeConfig` is what the
   binary hands in; the credential is read from a systemd credential and reaches the child on a
diff --git a/crates/claude-connector/Cargo.toml b/crates/claude-connector/Cargo.toml
index 7f9b042..6592252 100644
--- a/crates/claude-connector/Cargo.toml
+++ b/crates/claude-connector/Cargo.toml
@@ -13,6 +13,7 @@ path = "src/main.rs"
 
 [dependencies]
 anyhow = "1"
+hmac = "0.12"
 cultcache-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "a4eae9056152335662f1115f97bf35fb00ab6d23", package = "cultcache-rs" }
 model-connector = { path = "../model-connector", features = ["daemon"] }
 rmp-serde = "1"
diff --git a/crates/claude-connector/src/backend.rs b/crates/claude-connector/src/backend.rs
index 1963e1d..4833521 100644
--- a/crates/claude-connector/src/backend.rs
+++ b/crates/claude-connector/src/backend.rs
@@ -12,7 +12,9 @@ use sha2::{Digest, Sha256};
 use zeroize::Zeroizing;
 
 use crate::config::{ClaudeCodeConfig, PROVIDER};
-use crate::handle::{ClaudeCodeHandle, SessionStore, plan, prefix_digest, stdin_bytes};
+use crate::handle::{
+    Binding, ClaudeCodeHandle, MIN_SECRET_BYTES, SessionStore, plan, prefix_digest, stdin_bytes,
+};
 use crate::launch::{PromptFile, argv, child_env};
 use crate::process::{ProcessPort, RunningChild, StdoutRead};
 use crate::quota::QuotaCell;
@@ -36,6 +38,8 @@ pub trait TokenSource: Send + Sync {
 pub struct CredentialTokenSource {
     directory: Option<PathBuf>,
     name: String,
+    /// The configuration field `missing` names when the credential is absent.
+    field: &'static str,
 }
 
 impl CredentialTokenSource {
@@ -46,13 +50,21 @@ impl CredentialTokenSource {
         Self {
             directory: std::env::var_os("CREDENTIALS_DIRECTORY").map(PathBuf::from),
             name: name.into(),
+            field: "token_credential",
         }
     }
 
+    /// The same source, naming `field` when its credential is absent.
+    pub fn for_field(mut self, field: &'static str) -> Self {
+        self.field = field;
+        self
+    }
+
     pub fn in_directory(directory: impl Into<PathBuf>, name: impl Into<String>) -> Self {
         Self {
             directory: Some(directory.into()),
             name: name.into(),
+            field: "token_credential",
         }
     }
 }
@@ -72,7 +84,7 @@ impl TokenSource for CredentialTokenSource {
         if self.directory.is_none() {
             "credentials_directory"
         } else {
-            "token_credential"
+            self.field
         }
     }
 }
@@ -91,9 +103,17 @@ pub struct ClaudeCodeBackend {
     sessions: Box<dyn SessionStore>,
     sink: Box<dyn StderrSink>,
     tokens: Box<dyn TokenSource>,
+    handle_secret: Box<dyn TokenSource>,
     quota: Arc<QuotaCell>,
 }
 
+/// The handle secret, or `None` when it is absent or shorter than a key.
+fn usable_secret(source: &dyn TokenSource) -> Option<Zeroizing<String>> {
+    source
+        .read()
+        .filter(|secret| secret.len() >= MIN_SECRET_BYTES)
+}
+
 impl ClaudeCodeBackend {
     /// The backend over real processes and the CLI's real session store.
     #[cfg(unix)]
@@ -105,7 +125,11 @@ impl ClaudeCodeBackend {
         let tokens = Box::new(CredentialTokenSource::from_environment(
             config.token_credential.clone(),
         ));
-        Self::with_ports(config, port, sessions, sink, tokens)
+        let handle_secret = Box::new(
+            CredentialTokenSource::from_environment(config.handle_secret_credential.clone())
+                .for_field("handle_secret_credential"),
+        );
+        Self::with_ports(config, port, sessions, sink, tokens, handle_secret)
     }
 
     pub fn with_ports(
@@ -114,6 +138,7 @@ impl ClaudeCodeBackend {
         sessions: Box<dyn SessionStore>,
         sink: Box<dyn StderrSink>,
         tokens: Box<dyn TokenSource>,
+        handle_secret: Box<dyn TokenSource>,
     ) -> Self {
         Self {
             config,
@@ -121,6 +146,7 @@ impl ClaudeCodeBackend {
             sessions,
             sink,
             tokens,
+            handle_secret,
             quota: Arc::new(QuotaCell::new()),
         }
     }
@@ -136,8 +162,25 @@ impl ClaudeCodeBackend {
         binary_matches_pin(&self.config)
     }
 
-    fn run(&self, request: &ProviderRequest) -> (Vec<TransportEvent>, TransportOutcome) {
-        let plan = match plan(request, &self.config, self.sessions.as_ref()) {
+    fn run(
+        &self,
+        request: &ProviderRequest,
+        caller: &str,
+    ) -> (Vec<TransportEvent>, TransportOutcome) {
+        let Some(secret) = usable_secret(self.handle_secret.as_ref()) else {
+            return failed(
+                FailureKind::Authentication,
+                "claude-code: the connector has no handle secret".to_string(),
+                Some(zero_usage()),
+            );
+        };
+        let plan = match plan(
+            request,
+            &self.config,
+            secret.as_bytes(),
+            caller,
+            self.sessions.as_ref(),
+        ) {
             Ok(plan) => plan,
             Err(error) => {
                 return failed(
@@ -198,7 +241,7 @@ impl ClaudeCodeBackend {
         }
 
         match finished {
-            Ok(finished) => self.completed(request, finished),
+            Ok(finished) => self.completed(request, caller, secret.as_bytes(), finished),
             Err(failure_) => failure(failure_),
         }
     }
@@ -206,6 +249,8 @@ impl ClaudeCodeBackend {
     fn completed(
         &self,
         request: &ProviderRequest,
+        caller: &str,
+        secret: &[u8],
         finished: Finished,
     ) -> (Vec<TransportEvent>, TransportOutcome) {
         let mut payloads: Vec<TransportEventPayload> = finished
@@ -226,9 +271,19 @@ impl ClaudeCodeBackend {
         // and whatever follows it is seeded from the caller's transcript.
         if request.output_schema_json.is_none() {
             let handle = ClaudeCodeHandle::Ended {
-                session_id: finished.session_id,
+                session_id: finished.session_id.clone(),
                 instance: self.config.instance.clone(),
-                prefix_sha256: prefix_digest(&request.instructions, &request.model, &prefix),
+                prefix_sha256: prefix_digest(
+                    secret,
+                    &Binding {
+                        caller,
+                        instructions: &request.instructions,
+                        model: &request.model,
+                        session_id: &finished.session_id,
+                        instance: &self.config.instance,
+                        items: &prefix,
+                    },
+                ),
             };
             payloads.push(TransportEventPayload::ReasoningState {
                 state: handle.encode(),
@@ -256,7 +311,11 @@ impl ClaudeCodeBackend {
 
 impl ProviderBackend for ClaudeCodeBackend {
     fn ready(&self) -> Result<(), &'static str> {
-        readiness(&self.config, self.tokens.as_ref())
+        readiness(
+            &self.config,
+            self.tokens.as_ref(),
+            self.handle_secret.as_ref(),
+        )
     }
 
     fn provider(&self) -> &str {
@@ -264,7 +323,7 @@ impl ProviderBackend for ClaudeCodeBackend {
     }
 
     fn execute(&self, call: &ProviderCall) -> TransportResult {
-        let (events, outcome) = self.run(call.request());
+        let (events, outcome) = self.run(call.request(), call.caller_runtime_id());
         call.transported(events, outcome)
     }
 
@@ -327,6 +386,7 @@ fn binary_matches_pin(config: &ClaudeCodeConfig) -> bool {
 pub(crate) fn readiness(
     config: &ClaudeCodeConfig,
     tokens: &dyn TokenSource,
+    handle_secret: &dyn TokenSource,
 ) -> Result<(), &'static str> {
     if !config.work_dir.is_dir() {
         return Err("work_dir");
@@ -342,6 +402,9 @@ pub(crate) fn readiness(
     }
     // Last: outside the unit the credential is the one thing always absent,
     // and it must not hide the faults the document itself has.
+    if usable_secret(handle_secret).is_none() {
+        return Err(handle_secret.missing());
+    }
     if tokens.read().is_none() {
         return Err(tokens.missing());
     }
diff --git a/crates/claude-connector/src/backend/tests.rs b/crates/claude-connector/src/backend/tests.rs
index 2e3b993..fc2be99 100644
--- a/crates/claude-connector/src/backend/tests.rs
+++ b/crates/claude-connector/src/backend/tests.rs
@@ -333,6 +333,7 @@ fn config(dir: &Path, instance: &str, limits: ClaudeCodeLimits) -> ClaudeCodeCon
         config_dir: PathBuf::from("/cfg"),
         prompt_dir: dir.join("prompts"),
         token_credential: "claude-token".to_string(),
+        handle_secret_credential: "claude-handle-secret".to_string(),
         instance: instance.to_string(),
         limits,
     }
@@ -351,6 +352,26 @@ fn canary_token() -> Box<dyn TokenSource> {
     Box::new(FixedToken(Some(CANARY_TOKEN)))
 }
 
+/// A handle secret source with a fixed answer; it names its own field.
+struct FixedSecret(Option<&'static str>);
+
+impl TokenSource for FixedSecret {
+    fn read(&self) -> Option<Zeroizing<String>> {
+        self.0.map(|secret| Zeroizing::new(secret.to_string()))
+    }
+
+    fn missing(&self) -> &'static str {
+        "handle_secret_credential"
+    }
+}
+
+const SECRET: &str = "handle-secret-0000000000000000000000000000000000000000000000";
+const OTHER_SECRET: &str = "handle-secret-1111111111111111111111111111111111111111111111";
+
+fn canary_secret() -> Box<dyn TokenSource> {
+    Box::new(FixedSecret(Some(SECRET)))
+}
+
 fn limits() -> ClaudeCodeLimits {
     ClaudeCodeLimits {
         deadline: Duration::from_secs(30),
@@ -369,6 +390,16 @@ struct Rig {
 }
 
 fn rig_for(instance: &str, scripts: Vec<Script>, sessions: &[&str], stderr: &[u8]) -> Rig {
+    rig_with_secret(instance, scripts, sessions, stderr, Some(SECRET))
+}
+
+fn rig_with_secret(
+    instance: &str,
+    scripts: Vec<Script>,
+    sessions: &[&str],
+    stderr: &[u8],
+    secret: Option<&'static str>,
+) -> Rig {
     let spawns = Arc::new(AtomicUsize::new(0));
     let seen = Arc::new(Mutex::new(Seen::default()));
     let sink = Sink::default();
@@ -385,6 +416,7 @@ fn rig_for(instance: &str, scripts: Vec<Script>, sessions: &[&str], stderr: &[u8
         Box::new(Sessions(sessions.iter().map(|s| s.to_string()).collect())),
         Box::new(sink.clone()),
         canary_token(),
+        Box::new(FixedSecret(secret)),
     );
     Rig {
         dir,
@@ -401,8 +433,11 @@ fn rig(scripts: Vec<Script>) -> Rig {
 
 impl Rig {
     fn run(&self, request: ProviderRequest) -> TransportResult {
+        self.run_as(CALLER_1, request)
+    }
+    fn run_as(&self, caller: &str, request: ProviderRequest) -> TransportResult {
         let mut door = Door::new();
-        let (claim, _) = door.begin(request);
+        let (claim, _) = door.begin_as(caller, request);
         self.backend.execute(claim.call())
     }
     fn seen(&self) -> Seen {
@@ -417,10 +452,12 @@ impl Rig {
 
 static DOOR_SEQUENCE: AtomicUsize = AtomicUsize::new(0);
 
+const CALLER_1: &str = "caller-1";
+const CALLER_2: &str = "caller-2";
+
 struct Door {
     dir: PathBuf,
     service: TransportService,
-    key: TransportKey,
 }
 
 fn prices() -> PriceTable {
@@ -450,35 +487,42 @@ impl Door {
             DOOR_SEQUENCE.fetch_add(1, Ordering::SeqCst)
         ));
         std::fs::create_dir_all(&dir).unwrap();
-        let caller = CallerAdmission::new(
-            "caller-1",
-            "caller-1-key",
-            1,
-            [MODEL.to_string(), OTHER_MODEL.to_string()],
-            4,
-            FRAME / 2,
-            32_768,
-            SpendCeiling(1 << 40),
-            SpendWindowMs(3_600_000),
-        )
-        .unwrap();
+        let callers = [CALLER_1, CALLER_2].map(|id| {
+            CallerAdmission::new(
+                id,
+                format!("{id}-key"),
+                1,
+                [MODEL.to_string(), OTHER_MODEL.to_string()],
+                4,
+                FRAME / 2,
+                32_768,
+                SpendCeiling(1 << 40),
+                SpendWindowMs(3_600_000),
+            )
+            .unwrap()
+        });
         let service =
-            TransportService::open(&dir.join("replay.cc"), PROVIDER, prices(), [caller], 5_000)
+            TransportService::open(&dir.join("replay.cc"), PROVIDER, prices(), callers, 5_000)
                 .unwrap();
-        Self {
-            dir,
-            service,
-            key: TransportKey::from_connection_secret("caller-1-key").unwrap(),
-        }
+        Self { dir, service }
     }
 
     fn begin(
         &mut self,
         request: ProviderRequest,
     ) -> (model_connector::ExecutionClaim, TransportInvocation) {
+        self.begin_as(CALLER_1, request)
+    }
+
+    fn begin_as(
+        &mut self,
+        caller: &str,
+        request: ProviderRequest,
+    ) -> (model_connector::ExecutionClaim, TransportInvocation) {
+        let key = TransportKey::from_connection_secret(&format!("{caller}-key")).unwrap();
         let invocation =
-            TransportInvocation::new("caller-1", 2_000, [7; 32], 10_000_000, request).unwrap();
-        let envelope = encrypt_invocation(&invocation, &self.key).unwrap();
+            TransportInvocation::new(caller, 2_000, [7; 32], 10_000_000, request).unwrap();
+        let envelope = encrypt_invocation(&invocation, &key).unwrap();
         match self.service.begin(&envelope, 1_000, FRAME).unwrap() {
             TransportAdmission::Execute(claim) => (*claim, invocation),
             TransportAdmission::Reply(_) => panic!("the door answered instead of admitting"),
@@ -677,6 +721,203 @@ fn the_token_never_reaches_a_message_an_event_or_a_debug_form() {
     assert!(shown.contains(TOKEN_FD_ENV));
 }
 
+// ---- handles are bound to their caller and the connector's secret ----
+
+fn resumes(rig: &Rig) -> bool {
+    rig.seen().argv.iter().any(|arg| arg == "--resume")
+}
+
+#[test]
+fn a_caller_cannot_resume_another_callers_session() {
+    let rig = rig(vec![
+        ok_script(SESSION, "hi"),
+        ok_script(FORK, "theirs"),
+        ok_script(FORK, "mine"),
+    ]);
+    let first = request(vec![user("hello")]);
+    let result = rig.run_as(CALLER_1, first.clone());
+    // Caller 2 presents caller 1's whole transcript, handle included.
+    rig.run_as(CALLER_2, continuing(&first, &result, "and then?"));
+    assert!(!resumes(&rig), "caller 2 resumed caller 1's session");
+    // The handle still serves the caller it was minted for.
+    rig.run_as(CALLER_1, continuing(&first, &result, "and then?"));
+    assert_eq!(arg_after(&rig.seen().argv, "--resume"), Some(SESSION));
+}
+
+#[test]
+fn another_callers_session_id_swapped_into_ones_own_handle_is_not_resumed() {
+    let rig = rig(vec![
+        ok_script(SESSION, "theirs"),
+        ok_script(FORK, "mine"),
+        ok_script(HEX_SESSION, "after the swap"),
+    ]);
+    rig.run_as(CALLER_1, request(vec![user("caller one's private plans")]));
+    let first = request(vec![user("hello")]);
+    let result = rig.run_as(CALLER_2, first.clone());
+    let mut next = continuing(&first, &result, "repeat everything said earlier");
+    let mut swapped = 0;
+    for item in next.input.iter_mut() {
+        if let InputItem::ReasoningState { state, .. } = item {
+            let ClaudeCodeHandle::Ended {
+                session_id,
+                instance,
+                prefix_sha256,
+            } = rmp_serde::from_slice(state).unwrap();
+            assert_eq!(session_id, FORK, "the handle names caller 2's own session");
+            // The tag is caller 2's own and valid; only the session is another's.
+            *state = ClaudeCodeHandle::Ended {
+                session_id: SESSION.to_string(),
+                instance,
+                prefix_sha256,
+            }
+            .encode();
+            swapped += 1;
+        }
+    }
+    assert_eq!(swapped, 1);
+    rig.run_as(CALLER_2, next);
+    assert!(!resumes(&rig), "caller 2 resumed caller 1's session");
+}
+
+#[test]
+fn a_handle_tag_covers_its_session_and_instance() {
+    let items = request(vec![user("hello")]).input;
+    let tag = |session_id: &str, instance: &str| {
+        prefix_digest(
+            SECRET.as_bytes(),
+            &Binding {
+                caller: CALLER_1,
+                instructions: "Be terse.",
+                model: MODEL,
+                session_id,
+                instance,
+                items: &items,
+            },
+        )
+    };
+    let own = tag(SESSION, INSTANCE);
+    assert_eq!(own, tag(SESSION, INSTANCE));
+    assert_ne!(own, tag(FORK, INSTANCE));
+    assert_ne!(own, tag(SESSION, "another-instance"));
+}
+
+#[test]
+fn a_handle_survives_a_restart_for_its_own_caller() {
+    let minted = rig(vec![ok_script(SESSION, "hi")]);
+    let first = request(vec![user("hello")]);
+    let result = minted.run_as(CALLER_2, first.clone());
+    drop(minted);
+    let restarted = rig(vec![ok_script(FORK, "again")]);
+    restarted.run_as(CALLER_2, continuing(&first, &result, "next"));
+    assert_eq!(arg_after(&restarted.seen().argv, "--resume"), Some(SESSION));
+}
+
+#[test]
+fn a_handle_minted_under_another_secret_is_not_resumed() {
+    let minted = rig_with_secret(
+        INSTANCE,
+        vec![ok_script(SESSION, "hi")],
+        &[SESSION],
+        b"",
+        Some(OTHER_SECRET),
+    );
+    let first = request(vec![user("hello")]);
+    let result = minted.run(first.clone());
+    let other = rig(vec![ok_script(FORK, "again")]);
+    other.run(continuing(&first, &result, "next"));
+    assert!(!resumes(&other));
+}
+
+#[test]
+fn a_forged_handle_is_not_resumed() {
+    let first = request(vec![user("hello"), assistant("hi")]);
+    let forged = |tag: [u8; 32]| {
+        let handle = ClaudeCodeHandle::Ended {
+            session_id: SESSION.to_string(),
+            instance: INSTANCE.to_string(),
+            prefix_sha256: tag,
+        };
+        let mut input = first.input.clone();
+        input.push(InputItem::ReasoningState {
+            provider: PROVIDER.into(),
+            model: MODEL.into(),
+            state: handle.encode(),
+        });
+        input.push(user("next"));
+        request(input)
+    };
+    // No secret: the same digest, unkeyed, and a guess.
+    let bound = Binding {
+        caller: CALLER_1,
+        instructions: "Be terse.",
+        model: MODEL,
+        session_id: SESSION,
+        instance: INSTANCE,
+        items: &first.input,
+    };
+    let unkeyed = prefix_digest(b"", &bound);
+    let right_for_other_secret = prefix_digest(OTHER_SECRET.as_bytes(), &bound);
+    for tag in [unkeyed, right_for_other_secret, [0; 32]] {
+        let rig = rig(vec![ok_script(FORK, "x")]);
+        rig.run(forged(tag));
+        assert!(!resumes(&rig));
+    }
+    // The same request with the real tag resumes: the forgery failed on the tag alone.
+    let real = prefix_digest(SECRET.as_bytes(), &bound);
+    let rig = rig(vec![ok_script(FORK, "x")]);
+    rig.run(forged(real));
+    assert_eq!(arg_after(&rig.seen().argv, "--resume"), Some(SESSION));
+}
+
+#[test]
+fn no_usable_handle_secret_fails_before_spawn_and_is_not_ready() {
+    for secret in [None, Some("too-short")] {
+        let rig = rig_with_secret(
+            INSTANCE,
+            vec![ok_script(SESSION, "hi")],
+            &[SESSION],
+            b"",
+            secret,
+        );
+        let (kind, message, usage) = failure_of(&rig.run(request(vec![user("hello")])));
+        assert_eq!(kind, FailureKind::Authentication);
+        assert_eq!(usage, Some(usage_zero()));
+        assert!(!message.contains("too-short"));
+        assert_eq!(rig.spawns(), 0);
+    }
+}
+
+#[test]
+fn ready_wants_a_handle_secret_of_key_length() {
+    let dir = sandbox();
+    let (work, cfg) = (dir.path().join("work"), dir.path().join("cfg"));
+    std::fs::create_dir_all(&work).unwrap();
+    std::fs::create_dir_all(&cfg).unwrap();
+    for (secret, ready) in [
+        (None, Err("handle_secret_credential")),
+        (Some("too-short"), Err("handle_secret_credential")),
+        (Some(SECRET), Ok(())),
+    ] {
+        let mut config = config(dir.path(), INSTANCE, limits());
+        config.work_dir = work.clone();
+        config.config_dir = cfg.clone();
+        let backend = ClaudeCodeBackend::with_ports(
+            config,
+            Box::new(ScriptedPort {
+                scripts: Mutex::new(VecDeque::new()),
+                spawns: Arc::new(AtomicUsize::new(0)),
+                seen: Arc::new(Mutex::new(Seen::default())),
+                stderr: Vec::new(),
+            }),
+            Box::new(Sessions(vec![])),
+            Box::new(Sink::default()),
+            canary_token(),
+            Box::new(FixedSecret(secret)),
+        );
+        assert_eq!(backend.ready(), ready);
+    }
+}
+
 // ---- ready ----
 
 /// A backend over a scripted port that never has a script: `ready` runs nothing.
@@ -692,6 +933,7 @@ fn ready_backend(config: ClaudeCodeConfig, tokens: Box<dyn TokenSource>) -> Clau
         Box::new(Sessions(vec![])),
         Box::new(Sink::default()),
         tokens,
+        canary_secret(),
     )
 }
 
@@ -796,6 +1038,7 @@ fn token_comes_from_the_credential_file_not_the_environment() {
             &credentials,
             "claude-token",
         )),
+        canary_secret(),
     );
     let mut door = Door::new();
 
@@ -1464,6 +1707,7 @@ fn a_port_that_cannot_spawn_is_a_provider_failure() {
         Box::new(Sessions(vec![])),
         Box::new(Sink::default()),
         canary_token(),
+        canary_secret(),
     );
     let mut door = Door::new();
     let (claim, _) = door.begin(request(vec![user("hi")]));
@@ -1489,7 +1733,12 @@ fn backend_end_to_end_through_the_service() {
     let (claim, invocation) = door.begin(request(vec![user("hello")]));
     let result = rig.backend.execute(claim.call());
     let sealed = door.service.complete(claim, result, FRAME).unwrap();
-    let opened = decrypt_result(&sealed, &door.key, &invocation).unwrap();
+    let opened = decrypt_result(
+        &sealed,
+        &TransportKey::from_connection_secret("caller-1-key").unwrap(),
+        &invocation,
+    )
+    .unwrap();
     let (events, receipt) = parts(&opened);
     assert_eq!(receipt.provider, "claude-code");
     assert_eq!(receipt.model, MODEL);
@@ -2053,6 +2302,7 @@ mod real {
             Box::new(FsSessionStore::new(config.config_dir.clone())),
             Box::new(sink.clone()),
             canary_token(),
+            canary_secret(),
         );
         Fake {
             _guard: guard,
diff --git a/crates/claude-connector/src/config.rs b/crates/claude-connector/src/config.rs
index 3bd0e1f..9c0f239 100644
--- a/crates/claude-connector/src/config.rs
+++ b/crates/claude-connector/src/config.rs
@@ -25,6 +25,9 @@ pub struct ClaudeCodeConfig {
     /// The name of the systemd credential that holds the child's token, under
     /// `$CREDENTIALS_DIRECTORY`.
     pub token_credential: String,
+    /// The name of the systemd credential that holds the handle secret: the
+    /// key every handle is authenticated with.
+    pub handle_secret_credential: String,
     /// The connector instance id written into every handle.
     pub instance: String,
     pub limits: ClaudeCodeLimits,
diff --git a/crates/claude-connector/src/daemon.rs b/crates/claude-connector/src/daemon.rs
index 9f549fc..0e3c5e8 100644
--- a/crates/claude-connector/src/daemon.rs
+++ b/crates/claude-connector/src/daemon.rs
@@ -31,8 +31,8 @@ const MODEL_PREFIX: &str = "claude-";
 
 /// A unix account that may lease the subscription token for a Self run, the
 /// Self unit (`mind-self@<unit_instance>.service`) its process must run in,
-/// the caller whose spend window its leases reserve in, and what each lease
-/// reserves there.
+/// the configured caller whose spend window its leases reserve in, and what
+/// each lease reserves there.
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub struct LeaseCallerConfig {
     pub unix_user: String,
@@ -82,7 +82,9 @@ pub struct ClaudeConnectorConfig {
     pub socket_timeout_ms: u64,
     #[cultcache(key = 16)]
     pub max_expiry_skew_ms: u64,
-    /// Exactly one caller until claude-handle-caller-binding lands.
+    /// Every caller of this connector: its own connection key and epoch, models,
+    /// concurrency, payload and output bounds, and spend ceiling and window.
+    /// Ids are unique. Handles are bound to the caller that was given them.
     #[cultcache(key = 17)]
     pub callers: Vec<CallerConfig>,
     /// What each allowed model costs. The keys are the known list of models:
@@ -100,6 +102,10 @@ pub struct ClaudeConnectorConfig {
     /// The furthest a lease's expiry may lie from now.
     #[cultcache(key = 21)]
     pub lease_max_ms: u64,
+    /// The name of the systemd credential that holds the handle secret (at
+    /// least 32 bytes), the key every handle is authenticated with.
+    #[cultcache(key = 22)]
+    pub handle_secret_credential: String,
 }
 
 /// A refusal. It names the field or stage that failed and nothing the
@@ -180,6 +186,11 @@ impl ClaudeConnectorConfig {
         if !is_credential_name(&self.token_credential) {
             return Err(ConfigError::Invalid("token_credential"));
         }
+        if !is_credential_name(&self.handle_secret_credential)
+            || self.handle_secret_credential == self.token_credential
+        {
+            return Err(ConfigError::Invalid("handle_secret_credential"));
+        }
         if self.deadline_ms == 0
             || self.max_line_bytes == 0
             || self.max_total_bytes < self.max_line_bytes
@@ -194,37 +205,43 @@ impl ClaudeConnectorConfig {
         {
             return Err(ConfigError::Invalid("serve limits"));
         }
-        let [caller] = self.callers.as_slice() else {
+        if self.callers.is_empty() {
             return Err(ConfigError::Invalid("callers"));
-        };
-        if !is_trimmed_nonempty(&caller.caller_runtime_id) {
-            return Err(ConfigError::Invalid("caller_runtime_id"));
         }
-        if !caller.connection_key_file.is_absolute() {
-            return Err(ConfigError::Invalid("connection_key_file"));
-        }
-        if caller.connection_key_epoch == 0 {
-            return Err(ConfigError::Invalid("connection_key_epoch"));
-        }
-        if caller.max_concurrent_requests == 0
-            || caller.max_payload_bytes < MIN_PAYLOAD_BYTES
-            || caller.max_output_tokens == 0
-            || caller.spend_ceiling_micro_usd == 0
-            || caller.spend_window_ms == 0
-        {
-            return Err(ConfigError::Invalid("caller bounds"));
-        }
-        if caller.allowed_models.is_empty() {
-            return Err(ConfigError::Invalid("allowed_models"));
-        }
-        let mut seen = BTreeSet::new();
-        for model in &caller.allowed_models {
-            if !has_model_id_form(model)
-                || !seen.insert(model.as_str())
-                || !self.prices.contains_key(model)
+        let mut ids = BTreeSet::new();
+        for caller in &self.callers {
+            if !is_trimmed_nonempty(&caller.caller_runtime_id) {
+                return Err(ConfigError::Invalid("caller_runtime_id"));
+            }
+            if !ids.insert(caller.caller_runtime_id.as_str()) {
+                return Err(ConfigError::Invalid("callers"));
+            }
+            if !caller.connection_key_file.is_absolute() {
+                return Err(ConfigError::Invalid("connection_key_file"));
+            }
+            if caller.connection_key_epoch == 0 {
+                return Err(ConfigError::Invalid("connection_key_epoch"));
+            }
+            if caller.max_concurrent_requests == 0
+                || caller.max_payload_bytes < MIN_PAYLOAD_BYTES
+                || caller.max_output_tokens == 0
+                || caller.spend_ceiling_micro_usd == 0
+                || caller.spend_window_ms == 0
             {
+                return Err(ConfigError::Invalid("caller bounds"));
+            }
+            if caller.allowed_models.is_empty() {
                 return Err(ConfigError::Invalid("allowed_models"));
             }
+            let mut seen = BTreeSet::new();
+            for model in &caller.allowed_models {
+                if !has_model_id_form(model)
+                    || !seen.insert(model.as_str())
+                    || !self.prices.contains_key(model)
+                {
+                    return Err(ConfigError::Invalid("allowed_models"));
+                }
+            }
         }
         if !self
             .prices
@@ -233,11 +250,11 @@ impl ClaudeConnectorConfig {
         {
             return Err(ConfigError::Invalid("prices"));
         }
-        self.validate_leases(&caller.caller_runtime_id, caller.spend_ceiling_micro_usd)?;
+        self.validate_leases()?;
         Ok(listen)
     }
 
-    fn validate_leases(&self, caller_id: &str, spend_ceiling: u64) -> Result<(), ConfigError> {
+    fn validate_leases(&self) -> Result<(), ConfigError> {
         let Some(socket) = &self.lease_socket else {
             return if self.lease_callers.is_empty() {
                 Ok(())
@@ -255,9 +272,15 @@ impl ClaudeConnectorConfig {
         if self.lease_callers.is_empty()
             || !self.lease_callers.iter().all(|lease| {
                 is_trimmed_nonempty(&lease.unix_user)
-                    && lease.caller_id == caller_id
+                    && self
+                        .callers
+                        .iter()
+                        .find(|caller| caller.caller_runtime_id == lease.caller_id)
+                        .is_some_and(|caller| {
+                            (1..=caller.spend_ceiling_micro_usd)
+                                .contains(&lease.reservation_micro_usd)
+                        })
                     && is_unit_instance(&lease.unit_instance)
-                    && (1..=spend_ceiling).contains(&lease.reservation_micro_usd)
                     && users.insert(lease.unix_user.as_str())
             })
         {
@@ -274,6 +297,7 @@ impl ClaudeConnectorConfig {
             config_dir: self.config_dir.clone(),
             prompt_dir: self.prompt_dir.clone(),
             token_credential: self.token_credential.clone(),
+            handle_secret_credential: self.handle_secret_credential.clone(),
             instance: self.instance.clone(),
             limits: ClaudeCodeLimits {
                 deadline: Duration::from_millis(self.deadline_ms),
@@ -458,6 +482,8 @@ fn readiness_of(config: &ClaudeConnectorConfig) -> Result<Vec<CallerAdmission>,
     readiness(
         &backend,
         &CredentialTokenSource::from_environment(backend.token_credential.clone()),
+        &CredentialTokenSource::from_environment(backend.handle_secret_credential.clone())
+            .for_field("handle_secret_credential"),
     )
     .map_err(ConfigError::Unusable)?;
     if config.lease_socket.is_some() {
diff --git a/crates/claude-connector/src/handle.rs b/crates/claude-connector/src/handle.rs
index 222b74b..30cf6cf 100644
--- a/crates/claude-connector/src/handle.rs
+++ b/crates/claude-connector/src/handle.rs
@@ -5,9 +5,10 @@
 
 use std::path::PathBuf;
 
+use hmac::{Hmac, Mac};
 use model_connector::{InputItem, ProviderRequest};
 use serde::{Deserialize, Serialize};
-use sha2::{Digest, Sha256};
+use sha2::Sha256;
 
 use crate::config::{ClaudeCodeConfig, PROVIDER};
 
@@ -16,9 +17,11 @@ use crate::config::{ClaudeCodeConfig, PROVIDER};
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub(crate) enum ClaudeCodeHandle {
-    /// The session a finished turn left behind. `prefix_sha256` is the digest
-    /// of the instructions, the model and every transcript item that precedes
-    /// this handle.
+    /// The session a finished turn left behind. `prefix_sha256` is the
+    /// HMAC-SHA-256, keyed with the connector's handle secret, of the caller
+    /// that minted the handle, the instructions, the model, this session and
+    /// instance, and every transcript item that precedes this handle. Another caller, or anyone
+    /// without the secret, cannot produce or satisfy it.
     Ended {
         session_id: String,
         instance: String,
@@ -36,12 +39,43 @@ impl ClaudeCodeHandle {
     }
 }
 
-/// The digest a handle binds: named MessagePack of the instructions, the model
-/// and the transcript items before the handle.
-pub(crate) fn prefix_digest(instructions: &str, model: &str, items: &[InputItem]) -> [u8; 32] {
-    let bytes =
-        rmp_serde::to_vec_named(&(instructions, model, items)).expect("strings and bytes encode");
-    Sha256::digest(bytes).into()
+/// The smallest handle secret the connector accepts, in bytes.
+pub const MIN_SECRET_BYTES: usize = 32;
+
+/// Everything a handle's tag binds: the minting caller, the instructions, the
+/// model, the session and instance the handle names, and the transcript items
+/// before the handle. Swapping any of them into a valid handle breaks the tag.
+pub(crate) struct Binding<'a> {
+    pub(crate) caller: &'a str,
+    pub(crate) instructions: &'a str,
+    pub(crate) model: &'a str,
+    pub(crate) session_id: &'a str,
+    pub(crate) instance: &'a str,
+    pub(crate) items: &'a [InputItem],
+}
+
+fn prefix_mac(secret: &[u8], binding: &Binding<'_>) -> Hmac<Sha256> {
+    let bytes = rmp_serde::to_vec_named(&(
+        binding.caller,
+        binding.instructions,
+        binding.model,
+        binding.session_id,
+        binding.instance,
+        binding.items,
+    ))
+    .expect("strings and bytes encode");
+    let mut mac = Hmac::<Sha256>::new_from_slice(secret).expect("an HMAC key has any length");
+    mac.update(&bytes);
+    mac
+}
+
+/// The tag a handle carries: keyed named MessagePack of its [`Binding`].
+pub(crate) fn prefix_digest(secret: &[u8], binding: &Binding<'_>) -> [u8; 32] {
+    prefix_mac(secret, binding).finalize().into_bytes().into()
+}
+
+fn prefix_matches(secret: &[u8], binding: &Binding<'_>, tag: &[u8; 32]) -> bool {
+    prefix_mac(secret, binding).verify_slice(tag).is_ok()
 }
 
 /// Whether the CLI's session store still holds a session.
@@ -121,6 +155,8 @@ impl Plan {
 pub(crate) fn plan(
     request: &ProviderRequest,
     config: &ClaudeCodeConfig,
+    secret: &[u8],
+    caller: &str,
     sessions: &dyn SessionStore,
 ) -> Result<Plan, PlanError> {
     if !request.tools.is_empty() {
@@ -174,7 +210,18 @@ pub(crate) fn plan(
             .all(|item| matches!(item, InputItem::UserText { .. }));
         let resumable = instance == config.instance
             && tail_is_user_text
-            && prefix_sha256 == prefix_digest(&request.instructions, &request.model, &input[..at])
+            && prefix_matches(
+                secret,
+                &Binding {
+                    caller,
+                    instructions: &request.instructions,
+                    model: &request.model,
+                    session_id: &session_id,
+                    instance: &instance,
+                    items: &input[..at],
+                },
+                &prefix_sha256,
+            )
             && sessions.exists(&session_id);
         if resumable {
             return Ok(Plan::Resume {
diff --git a/crates/claude-connector/src/lib.rs b/crates/claude-connector/src/lib.rs
index 22ff46f..6ed8757 100644
--- a/crates/claude-connector/src/lib.rs
+++ b/crates/claude-connector/src/lib.rs
@@ -23,7 +23,7 @@ pub use daemon::{
 };
 #[cfg(unix)]
 pub use daemon::{check_prompt_dir, serve_config};
-pub use handle::{FsSessionStore, SessionStore};
+pub use handle::{FsSessionStore, MIN_SECRET_BYTES, SessionStore};
 pub use launch::ChildEnv;
 #[cfg(target_os = "linux")]
 pub use lease::{
diff --git a/crates/claude-connector/src/main.rs b/crates/claude-connector/src/main.rs
index 0ea296f..e7f28bc 100644
--- a/crates/claude-connector/src/main.rs
+++ b/crates/claude-connector/src/main.rs
@@ -2,6 +2,8 @@
 //!
 //!   claude-connector serve --config PATH.cc
 //!   claude-connector config init PATH.cc --flag value ...
+//!   claude-connector config add-caller PATH.cc --caller-id ID --connection-key-file F ...
+//!   claude-connector config add-lease-caller PATH.cc --lease-user U --caller-id ID ...
 //!   claude-connector config check PATH.cc
 //!
 //! `serve` takes its configuration from the typed document alone. The flags of
@@ -13,7 +15,8 @@ use std::path::PathBuf;
 use std::process::ExitCode;
 
 use claude_connector::{
-    ClaudeConnectorConfig, ConfigError, LeaseCallerConfig, check_ready, load_config, write_config,
+    ClaudeConnectorConfig, ConfigError, LeaseCallerConfig, MIN_SECRET_BYTES, check_ready,
+    load_config, write_config,
 };
 use model_connector::{CallerConfig, ModelPrice, Rates, frame_bytes_for_payload};
 
@@ -23,6 +26,8 @@ fn main() -> ExitCode {
     let result = match args.as_slice() {
         ["serve", "--config", path] => serve(path),
         ["config", "init", path, flags @ ..] => init(path, flags),
+        ["config", "add-caller", path, flags @ ..] => add_caller(path, flags),
+        ["config", "add-lease-caller", path, flags @ ..] => add_lease_caller(path, flags),
         ["config", "check", path] => check(path),
         _ => Err(Failure::Usage),
     };
@@ -38,7 +43,7 @@ fn main() -> ExitCode {
 #[derive(Debug, thiserror::Error)]
 enum Failure {
     #[error(
-        "usage: serve --config PATH | config init PATH --flag value ... | config check PATH (check reads the credential from the environment it runs in: run it in the unit's environment)"
+        "usage: serve --config PATH | config init PATH --flag value ... | config add-caller PATH --flag value ... | config add-lease-caller PATH --flag value ... | config check PATH (check reads the credentials from the environment it runs in: run it in the unit's environment)"
     )]
     Usage,
     #[error(transparent)]
@@ -66,20 +71,31 @@ fn check(path: &str) -> Result<(), Failure> {
     check_ready(&config)?;
     println!("claude connector configuration is valid and ready to serve");
     println!("callers: {}", config.callers.len());
-    println!("allowed_models: {}", config.callers[0].allowed_models.len());
+    println!("lease_callers: {}", config.lease_callers.len());
     println!("prices: {}", config.prices.len());
     Ok(())
 }
 
-/// The flags of `config init`, each given at most once except the two lists.
+/// The flags of one config verb: each single flag at most once, each list
+/// flag any number of times.
 struct Flags {
     single: BTreeMap<&'static str, String>,
-    models: Vec<String>,
-    prices: Vec<String>,
-    lease_users: Vec<String>,
+    lists: BTreeMap<&'static str, Vec<String>>,
 }
 
-const SINGLE: [&str; 26] = [
+/// The flags that describe one caller, shared by `config init` (which writes
+/// the first caller) and `config add-caller`.
+const CALLER: [&str; 7] = [
+    "--caller-id",
+    "--connection-key-file",
+    "--connection-key-epoch",
+    "--spend-ceiling-micro-usd",
+    "--spend-window-ms",
+    "--max-concurrent-requests",
+    "--max-output-tokens",
+];
+
+const INIT: [&str; 21] = [
     "--listen",
     "--replay-store",
     "--claude-binary",
@@ -89,18 +105,13 @@ const SINGLE: [&str; 26] = [
     "--prompt-dir",
     "--instance",
     "--token-credential",
-    "--caller-id",
-    "--connection-key-file",
-    "--connection-key-epoch",
-    "--spend-ceiling-micro-usd",
-    "--spend-window-ms",
+    "--handle-secret-credential",
+    "--handle-secret-file",
     "--deadline-ms",
     "--max-line-bytes",
     "--max-total-bytes",
     "--stderr-tail-bytes",
-    "--max-concurrent-requests",
     "--max-payload-bytes",
-    "--max-output-tokens",
     "--socket-timeout-ms",
     "--lease-socket",
     "--lease-max-ms",
@@ -108,36 +119,53 @@ const SINGLE: [&str; 26] = [
     "--lease-reservation-micro-usd",
 ];
 
+const LEASE_CALLER: [&str; 3] = [
+    "--caller-id",
+    "--lease-unit-instance",
+    "--lease-reservation-micro-usd",
+];
+
 impl Flags {
-    fn parse(args: &[&str]) -> Result<Self, Failure> {
+    fn parse(
+        args: &[&str],
+        singles: &[&[&'static str]],
+        lists: &[&'static str],
+    ) -> Result<Self, Failure> {
         let mut flags = Flags {
             single: BTreeMap::new(),
-            models: Vec::new(),
-            prices: Vec::new(),
-            lease_users: Vec::new(),
+            lists: BTreeMap::new(),
         };
         let mut args = args.iter();
         while let Some(&name) = args.next() {
             let Some(&value) = args.next() else {
                 return Err(Failure::Flags);
             };
-            match name {
-                "--allowed-model" => flags.models.push(value.to_string()),
-                "--price" => flags.prices.push(value.to_string()),
-                "--lease-user" => flags.lease_users.push(value.to_string()),
-                _ => {
-                    let Some(known) = SINGLE.iter().find(|known| **known == name) else {
-                        return Err(Failure::Flags);
-                    };
-                    if flags.single.insert(*known, value.to_string()).is_some() {
-                        return Err(Failure::Flags);
-                    }
-                }
+            if let Some(list) = lists.iter().find(|known| **known == name) {
+                flags
+                    .lists
+                    .entry(*list)
+                    .or_default()
+                    .push(value.to_string());
+                continue;
+            }
+            let Some(known) = singles
+                .iter()
+                .flat_map(|set| set.iter())
+                .find(|k| **k == name)
+            else {
+                return Err(Failure::Flags);
+            };
+            if flags.single.insert(*known, value.to_string()).is_some() {
+                return Err(Failure::Flags);
             }
         }
         Ok(flags)
     }
 
+    fn list(&self, name: &'static str) -> &[String] {
+        self.lists.get(name).map_or(&[], Vec::as_slice)
+    }
+
     fn text(&self, name: &'static str) -> Result<String, Failure> {
         self.single.get(name).cloned().ok_or(Failure::Missing(name))
     }
@@ -162,6 +190,22 @@ impl Flags {
             Some(text) => text.parse().map_err(|_| Failure::Malformed(name)),
         }
     }
+
+    /// The caller these flags describe. Its payload bound comes from
+    /// `--max-payload-bytes` when the verb takes one.
+    fn caller(&self) -> Result<CallerConfig, Failure> {
+        Ok(CallerConfig {
+            caller_runtime_id: self.text("--caller-id")?,
+            connection_key_file: self.path("--connection-key-file")?,
+            connection_key_epoch: self.number("--connection-key-epoch")?,
+            allowed_models: self.list("--allowed-model").to_vec(),
+            max_concurrent_requests: self.number_or("--max-concurrent-requests", 4)?,
+            max_payload_bytes: self.number_or("--max-payload-bytes", 1 << 20)?,
+            max_output_tokens: self.number_or("--max-output-tokens", 32_768)?,
+            spend_ceiling_micro_usd: self.number("--spend-ceiling-micro-usd")?,
+            spend_window_ms: self.number("--spend-window-ms")?,
+        })
+    }
 }
 
 fn sha256(hex: &str) -> Option<[u8; 32]> {
@@ -202,30 +246,94 @@ fn price(text: &str) -> Option<(String, ModelPrice)> {
     ))
 }
 
+/// Writes 32 random bytes as 64 hex characters, mode 0600, unless the file is
+/// already there: the secret is never read back or shown, and a file that
+/// exists is the secret handles already depend on, so it is kept as it is when
+/// it is a regular file, mode 0600, owned by this user and at least
+/// `MIN_SECRET_BYTES` long, and refused untouched otherwise.
+#[cfg(unix)]
+fn ensure_secret_file(path: &std::path::Path) -> Result<(), Failure> {
+    use std::io::{Read, Write};
+    use std::os::unix::fs::OpenOptionsExt;
+
+    if !path.is_absolute() {
+        return Err(Failure::Malformed("--handle-secret-file"));
+    }
+    let mut file = match std::fs::OpenOptions::new()
+        .write(true)
+        .create_new(true)
+        .mode(0o600)
+        .open(path)
+    {
+        Ok(file) => file,
+        Err(_) => return existing_secret_is_usable(path),
+    };
+    let mut bytes = [0_u8; 32];
+    std::fs::File::open("/dev/urandom")
+        .and_then(|mut random| random.read_exact(&mut bytes))
+        .map_err(|_| Failure::Malformed("--handle-secret-file"))?;
+    let hex: String = bytes.iter().map(|byte| format!("{byte:02x}")).collect();
+    file.write_all(hex.as_bytes())
+        .map_err(|_| Failure::Malformed("--handle-secret-file"))
+}
+
+/// An existing secret file is accepted only as the connector would use it: not
+/// a link, a regular file, readable by no one else, ours and long enough.
+#[cfg(unix)]
+fn existing_secret_is_usable(path: &std::path::Path) -> Result<(), Failure> {
+    let metadata =
+        std::fs::symlink_metadata(path).map_err(|_| Failure::Malformed("--handle-secret-file"))?;
+    // SAFETY: geteuid has no preconditions and cannot fail.
+    let uid = unsafe { libc::geteuid() };
+    if secret_file_usable(&metadata, uid) {
+        Ok(())
+    } else {
+        Err(Failure::Malformed("--handle-secret-file"))
+    }
+}
+
+#[cfg(unix)]
+fn secret_file_usable(metadata: &std::fs::Metadata, uid: u32) -> bool {
+    use std::os::unix::fs::{MetadataExt, PermissionsExt};
+
+    metadata.is_file()
+        && metadata.permissions().mode() & 0o7777 == 0o600
+        && metadata.uid() == uid
+        && metadata.len() >= MIN_SECRET_BYTES as u64
+}
+
+#[cfg(not(unix))]
+fn ensure_secret_file(_: &std::path::Path) -> Result<(), Failure> {
+    Err(Failure::Usage)
+}
+
 fn init(path: &str, args: &[&str]) -> Result<(), Failure> {
-    let flags = Flags::parse(args)?;
-    let max_payload_bytes: usize = flags.number_or("--max-payload-bytes", 1 << 20)?;
-    let max_concurrent_requests: usize = flags.number_or("--max-concurrent-requests", 4)?;
+    let flags = Flags::parse(
+        args,
+        &[&INIT, &CALLER],
+        &["--allowed-model", "--price", "--lease-user"],
+    )?;
+    let first = flags.caller()?;
     let mut prices = BTreeMap::new();
-    for text in &flags.prices {
+    for text in flags.list("--price") {
         let (model, price) = price(text).ok_or(Failure::Malformed("--price"))?;
         if prices.insert(model, price).is_some() {
             return Err(Failure::Malformed("--price"));
         }
     }
-    let caller_id = flags.text("--caller-id")?;
     let lease_callers = flags
-        .lease_users
+        .list("--lease-user")
         .iter()
         .map(|unix_user| {
             Ok(LeaseCallerConfig {
                 unix_user: unix_user.clone(),
-                caller_id: caller_id.clone(),
+                caller_id: first.caller_runtime_id.clone(),
                 unit_instance: flags.text("--lease-unit-instance")?,
                 reservation_micro_usd: flags.number("--lease-reservation-micro-usd")?,
             })
         })
         .collect::<Result<Vec<_>, Failure>>()?;
+    let secret_file = flags.path("--handle-secret-file")?;
     let config = ClaudeConnectorConfig {
         listen: flags.text("--listen")?,
         replay_store: flags.path("--replay-store")?,
@@ -241,26 +349,80 @@ fn init(path: &str, args: &[&str]) -> Result<(), Failure> {
         max_line_bytes: flags.number_or("--max-line-bytes", 4 << 20)?,
         max_total_bytes: flags.number_or("--max-total-bytes", 64 << 20)?,
         stderr_tail_bytes: flags.number_or("--stderr-tail-bytes", 8192)?,
-        max_frame_bytes: frame_bytes_for_payload(max_payload_bytes),
-        max_connections: max_concurrent_requests.max(8),
+        max_frame_bytes: frame_bytes_for_payload(first.max_payload_bytes),
+        max_connections: first.max_concurrent_requests.max(8),
         socket_timeout_ms: flags.number_or("--socket-timeout-ms", 10_000)?,
         max_expiry_skew_ms: 300_000,
-        callers: vec![CallerConfig {
-            caller_runtime_id: caller_id.clone(),
-            connection_key_file: flags.path("--connection-key-file")?,
-            connection_key_epoch: flags.number("--connection-key-epoch")?,
-            allowed_models: flags.models.clone(),
-            max_concurrent_requests,
-            max_payload_bytes,
-            max_output_tokens: flags.number_or("--max-output-tokens", 32_768)?,
-            spend_ceiling_micro_usd: flags.number("--spend-ceiling-micro-usd")?,
-            spend_window_ms: flags.number("--spend-window-ms")?,
-        }],
+        callers: vec![first],
         prices,
         lease_socket: flags.single.get("--lease-socket").map(PathBuf::from),
         lease_callers,
         lease_max_ms: flags.number_or("--lease-max-ms", 3_600_000)?,
+        handle_secret_credential: flags.text("--handle-secret-credential")?,
     };
+    // Validate before the secret file exists: a refused document leaves nothing.
+    config.validate()?;
+    ensure_secret_file(&secret_file)?;
     write_config(path.as_ref(), &config)?;
     Ok(())
 }
+
+/// Appends one caller to an existing document. The serve limits grow to carry
+/// it: the frame to its payload bound, the connections to every caller's slots.
+fn add_caller(path: &str, args: &[&str]) -> Result<(), Failure> {
+    let flags = Flags::parse(
+        args,
+        &[&CALLER, &["--max-payload-bytes"]],
+        &["--allowed-model"],
+    )?;
+    let caller = flags.caller()?;
+    let mut config = load_config(path.as_ref())?;
+    config.max_frame_bytes = config
+        .max_frame_bytes
+        .max(frame_bytes_for_payload(caller.max_payload_bytes));
+    config.callers.push(caller);
+    let slots: usize = config
+        .callers
+        .iter()
+        .map(|caller| caller.max_concurrent_requests)
+        .sum();
+    config.max_connections = config.max_connections.max(slots);
+    write_config(path.as_ref(), &config)?;
+    Ok(())
+}
+
+/// Appends one leasing account to an existing document that lends a token.
+fn add_lease_caller(path: &str, args: &[&str]) -> Result<(), Failure> {
+    let flags = Flags::parse(args, &[&LEASE_CALLER, &["--lease-user"]], &[])?;
+    let mut config = load_config(path.as_ref())?;
+    config.lease_callers.push(LeaseCallerConfig {
+        unix_user: flags.text("--lease-user")?,
+        caller_id: flags.text("--caller-id")?,
+        unit_instance: flags.text("--lease-unit-instance")?,
+        reservation_micro_usd: flags.number("--lease-reservation-micro-usd")?,
+    });
+    write_config(path.as_ref(), &config)?;
+    Ok(())
+}
+
+#[cfg(all(test, unix))]
+mod tests {
+    use std::os::unix::fs::{MetadataExt, PermissionsExt};
+
+    use super::*;
+
+    #[test]
+    fn a_secret_file_owned_by_someone_else_is_not_usable() {
+        let dir =
+            std::env::temp_dir().join(format!("claude-connector-main-{}", std::process::id()));
+        std::fs::create_dir_all(&dir).unwrap();
+        let path = dir.join("handle.secret");
+        std::fs::write(&path, "k".repeat(64)).unwrap();
+        std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600)).unwrap();
+        let metadata = std::fs::symlink_metadata(&path).unwrap();
+        let owner = metadata.uid();
+        assert!(secret_file_usable(&metadata, owner));
+        assert!(!secret_file_usable(&metadata, owner.wrapping_add(1)));
+        std::fs::remove_dir_all(&dir).unwrap();
+    }
+}
diff --git a/crates/claude-connector/src/quota.rs b/crates/claude-connector/src/quota.rs
index 76f0ce4..e4ea4b3 100644
--- a/crates/claude-connector/src/quota.rs
+++ b/crates/claude-connector/src/quota.rs
@@ -76,7 +76,7 @@ fn unix_ms() -> u64 {
 mod tests {
     use super::*;
 
-    const RESET: u64 = 1_793_307_000_000;
+    const RESET: u64 = 4_102_444_800_000;
 
     fn quota(window: &str, permille: u32, observed_at: u64) -> PlanQuota {
         PlanQuota {
diff --git a/crates/claude-connector/tests/config_document.rs b/crates/claude-connector/tests/config_document.rs
index 80285a2..3659113 100644
--- a/crates/claude-connector/tests/config_document.rs
+++ b/crates/claude-connector/tests/config_document.rs
@@ -74,6 +74,7 @@ fn valid() -> ClaudeConnectorConfig {
         prompt_dir: "/var/lib/claude-connector/prompts".into(),
         instance: "persona-a".to_string(),
         token_credential: "claude-token".to_string(),
+        handle_secret_credential: "claude-handle-secret".to_string(),
         deadline_ms: 600_000,
         max_line_bytes: 4 << 20,
         max_total_bytes: 64 << 20,
@@ -326,11 +327,10 @@ fn config_refuses_aliases_unpriced_models_and_second_callers() {
             Box::new(|c| c.callers[0].allowed_models.clear()),
         ),
         (
-            "second caller",
+            "duplicate caller id",
             Box::new(|c| {
-                let mut second = c.callers[0].clone();
-                second.caller_runtime_id = "other".to_string();
-                c.callers.push(second);
+                let again = c.callers[0].clone();
+                c.callers.push(again);
             }),
         ),
         ("no caller", Box::new(|c| c.callers.clear())),
@@ -538,7 +538,7 @@ fn no_refusal_echoes_a_value() {
         }),
         Box::new(|c| {
             let mut second = c.callers[0].clone();
-            second.caller_runtime_id = CANARY.to_string();
+            second.caller_runtime_id = format!(" {CANARY}");
             c.callers.push(second);
         }),
     ];
@@ -646,6 +646,10 @@ fn init_flags<'a>(scratch: &'a Scratch, extra: &[&'a str]) -> Vec<String> {
         "persona-a",
         "--token-credential",
         "claude-token",
+        "--handle-secret-credential",
+        "claude-handle-secret",
+        "--handle-secret-file",
+        &path("handle.secret"),
         "--caller-id",
         "persona-a-caller",
         "--connection-key-file",
@@ -704,6 +708,11 @@ fn prepare(scratch: &Scratch) -> (Vec<String>, PathBuf) {
     )
     .unwrap();
     std::fs::write(scratch.path("credentials/claude-token"), CANARY).unwrap();
+    std::fs::write(
+        scratch.path("credentials/claude-handle-secret"),
+        "s".repeat(64),
+    )
+    .unwrap();
     std::fs::write(scratch.path("caller.key"), "caller-secret").unwrap();
     std::fs::write(scratch.path("claude"), b"pinned claude").unwrap();
     let sha: String = Sha256::digest(b"pinned claude")
@@ -749,7 +758,7 @@ fn check_passes_a_document_serve_can_start_and_reports_counts_only() {
     );
     assert!(checked.status.success(), "{}", shown(&checked));
     let text = shown(&checked);
-    assert!(text.contains("callers: 1") && text.contains("allowed_models: 1"));
+    assert!(text.contains("callers: 1") && text.contains("lease_callers: 0"));
     assert!(text.contains("prices: 1"));
     for secret in ["persona-a", HAIKU, CANARY, "caller-secret"] {
         assert!(!text.contains(secret), "{text}");
@@ -788,6 +797,32 @@ fn check_refuses_exactly_what_serve_refuses_and_names_the_field() {
             config.max_frame_bytes = 4099;
             write_config(&path, &config).unwrap();
         }),
+        ("two callers on one key", "callers", |s, _| {
+            std::fs::write(s.path("twin.key"), "caller-secret").unwrap();
+            let config = s.path("connector.cc");
+            let twin = s.path("twin.key");
+            let added = run(
+                &[
+                    "config",
+                    "add-caller",
+                    config.to_str().unwrap(),
+                    "--caller-id",
+                    "twin",
+                    "--connection-key-file",
+                    twin.to_str().unwrap(),
+                    "--connection-key-epoch",
+                    "1",
+                    "--allowed-model",
+                    HAIKU,
+                    "--spend-ceiling-micro-usd",
+                    "1000",
+                    "--spend-window-ms",
+                    "60000",
+                ],
+                &[],
+            );
+            assert!(added.status.success(), "{}", shown(&added));
+        }),
         ("too few connections", "connection capacity", |s, _| {
             let path = s.path("connector.cc");
             let mut config = load_config(&path).unwrap();
@@ -1206,3 +1241,430 @@ fn serve_binds_the_lease_socket_and_answers_a_stranger_with_a_refusal() {
     outcome.unwrap();
     assert!(running);
 }
+
+// ---- many callers ----
+
+fn second_caller() -> CallerConfig {
+    CallerConfig {
+        caller_runtime_id: "heartbeat".to_string(),
+        connection_key_file: "/etc/claude-connector/heartbeat.key".into(),
+        connection_key_epoch: 3,
+        allowed_models: vec![HAIKU.to_string()],
+        max_concurrent_requests: 1,
+        max_payload_bytes: 64 << 10,
+        max_output_tokens: 1024,
+        spend_ceiling_micro_usd: 1_000_000,
+        spend_window_ms: 60_000,
+    }
+}
+
+#[test]
+fn several_callers_round_trip_and_each_is_validated_alone() {
+    let scratch = Scratch::new();
+    let path = scratch.path("connector.cc");
+    let mut config = valid();
+    config.callers.push(second_caller());
+    write_config(&path, &config).unwrap();
+    assert_eq!(load_config(&path).unwrap(), config);
+
+    let broken: Vec<(&str, Change)> = vec![
+        (
+            "caller bounds",
+            Box::new(|c| c.callers[1].spend_ceiling_micro_usd = 0),
+        ),
+        (
+            "connection_key_epoch",
+            Box::new(|c| c.callers[1].connection_key_epoch = 0),
+        ),
+        (
+            "connection_key_file",
+            Box::new(|c| c.callers[1].connection_key_file = "heartbeat.key".into()),
+        ),
+        (
+            "allowed_models",
+            Box::new(|c| c.callers[1].allowed_models = vec!["claude-opus-9".to_string()]),
+        ),
+        (
+            "callers",
+            Box::new(|c| c.callers[1].caller_runtime_id = c.callers[0].caller_runtime_id.clone()),
+        ),
+    ];
+    for (field, change) in &broken {
+        let mut config = valid();
+        config.callers.push(second_caller());
+        change(&mut config);
+        assert_eq!(refused(&config), *field);
+    }
+}
+
+#[test]
+fn a_lease_caller_bills_one_configured_caller_within_that_callers_ceiling() {
+    let mut config = valid();
+    config.callers.push(second_caller());
+    let lease = |user: &str, caller: &str, instance: &str, reservation: u64| LeaseCallerConfig {
+        unix_user: user.to_string(),
+        caller_id: caller.to_string(),
+        unit_instance: instance.to_string(),
+        reservation_micro_usd: reservation,
+    };
+    // A lease caller of each caller, each within its own caller's ceiling.
+    config
+        .lease_callers
+        .push(lease("mind-heartbeat", "heartbeat", "heartbeat", 1_000_000));
+    config.validate().unwrap();
+    // The first caller's ceiling does not stretch to the second's lease, nor the reverse.
+    config.lease_callers[1].reservation_micro_usd = 1_000_001;
+    assert_eq!(refused(&config), "lease_callers");
+    config.lease_callers[1].reservation_micro_usd = 1;
+    config.lease_callers[0].reservation_micro_usd = 50_000_001;
+    assert_eq!(refused(&config), "lease_callers");
+    config.lease_callers[0].reservation_micro_usd = 50_000_000;
+    // A lease caller naming no configured caller, or sharing another's account.
+    config.lease_callers[1].caller_id = "nobody".to_string();
+    assert_eq!(refused(&config), "lease_callers");
+    config.lease_callers[1].caller_id = "heartbeat".to_string();
+    config.lease_callers[1].unix_user = config.lease_callers[0].unix_user.clone();
+    assert_eq!(refused(&config), "lease_callers");
+}
+
+#[test]
+fn add_caller_and_add_lease_caller_extend_the_document_and_grow_the_serve_limits() {
+    let scratch = Scratch::new();
+    let made = init(
+        &scratch,
+        &init_flags(
+            &scratch,
+            &[
+                "--lease-socket",
+                "/run/claude-connector/lease.sock",
+                "--lease-user",
+                "mind-eureka-self",
+                "--lease-unit-instance",
+                "eureka",
+                "--lease-reservation-micro-usd",
+                "5000000",
+            ],
+        ),
+    );
+    assert!(made.status.success(), "{}", shown(&made));
+    let path = scratch.path("connector.cc");
+    let path_text = path.to_str().unwrap();
+    let before = load_config(&path).unwrap();
+
+    let key = scratch.path("heartbeat.key");
+    let added = run(
+        &[
+            "config",
+            "add-caller",
+            path_text,
+            "--caller-id",
+            "heartbeat",
+            "--connection-key-file",
+            key.to_str().unwrap(),
+            "--connection-key-epoch",
+            "2",
+            "--allowed-model",
+            HAIKU,
+            "--spend-ceiling-micro-usd",
+            "2000000",
+            "--spend-window-ms",
+            "60000",
+            "--max-concurrent-requests",
+            "30",
+            "--max-payload-bytes",
+            "4194304",
+        ],
+        &[],
+    );
+    assert!(added.status.success(), "{}", shown(&added));
+    let leased = run(
+        &[
+            "config",
+            "add-lease-caller",
+            path_text,
+            "--lease-user",
+            "mind-heartbeat",
+            "--caller-id",
+            "heartbeat",
+            "--lease-unit-instance",
+            "heartbeat",
+            "--lease-reservation-micro-usd",
+            "1000000",
+        ],
+        &[],
+    );
+    assert!(leased.status.success(), "{}", shown(&leased));
+
+    let after = load_config(&path).unwrap();
+    assert_eq!(after.callers[0], before.callers[0]);
+    assert_eq!(after.callers.len(), 2);
+    let heartbeat = &after.callers[1];
+    assert_eq!(heartbeat.caller_runtime_id, "heartbeat");
+    assert_eq!(heartbeat.connection_key_epoch, 2);
+    assert_eq!(heartbeat.spend_ceiling_micro_usd, 2_000_000);
+    assert_eq!(heartbeat.max_payload_bytes, 4 << 20);
+    assert_eq!(after.lease_callers[0], before.lease_callers[0]);
+    assert_eq!(after.lease_callers[1].caller_id, "heartbeat");
+    // The frame carries the largest payload and the connections every slot.
+    assert!(after.max_frame_bytes > before.max_frame_bytes);
+    assert!(after.max_connections >= 34);
+    assert_eq!(after.prices, before.prices);
+    assert_eq!(
+        after.handle_secret_credential,
+        before.handle_secret_credential
+    );
+
+    // Repeating either verb is refused and leaves the document as it was.
+    let again = run(
+        &[
+            "config",
+            "add-caller",
+            path_text,
+            "--caller-id",
+            "heartbeat",
+            "--connection-key-file",
+            key.to_str().unwrap(),
+            "--connection-key-epoch",
+            "2",
+            "--allowed-model",
+            HAIKU,
+            "--spend-ceiling-micro-usd",
+            "2000000",
+            "--spend-window-ms",
+            "60000",
+        ],
+        &[],
+    );
+    assert!(!again.status.success());
+    assert_eq!(load_config(&path).unwrap(), after);
+    let unleased = run(
+        &[
+            "config",
+            "add-lease-caller",
+            path_text,
+            "--lease-user",
+            "mind-heartbeat",
+            "--caller-id",
+            "heartbeat",
+            "--lease-unit-instance",
+            "heartbeat",
+            "--lease-reservation-micro-usd",
+            "1000000",
+        ],
+        &[],
+    );
+    assert!(!unleased.status.success());
+    assert_eq!(load_config(&path).unwrap(), after);
+}
+
+#[test]
+fn add_lease_caller_needs_a_document_that_lends_a_token() {
+    let scratch = Scratch::new();
+    let made = init(&scratch, &init_flags(&scratch, &[]));
+    assert!(made.status.success(), "{}", shown(&made));
+    let path = scratch.path("connector.cc");
+    let leased = run(
+        &[
+            "config",
+            "add-lease-caller",
+            path.to_str().unwrap(),
+            "--lease-user",
+            "mind-heartbeat",
+            "--caller-id",
+            "persona-a-caller",
+            "--lease-unit-instance",
+            "heartbeat",
+            "--lease-reservation-micro-usd",
+            "1000000",
+        ],
+        &[],
+    );
+    assert!(!leased.status.success());
+    assert!(load_config(&path).unwrap().lease_callers.is_empty());
+}
+
+#[cfg(unix)]
+#[test]
+fn init_creates_the_handle_secret_once_private_and_never_shows_it() {
+    use std::os::unix::fs::PermissionsExt;
+
+    let scratch = Scratch::new();
+    let secret = scratch.path("handle.secret");
+    let made = init(&scratch, &init_flags(&scratch, &[]));
+    assert!(made.status.success(), "{}", shown(&made));
+    let metadata = std::fs::metadata(&secret).unwrap();
+    assert_eq!(metadata.permissions().mode() & 0o777, 0o600);
+    let text = std::fs::read_to_string(&secret).unwrap();
+    assert_eq!(text.len(), 64);
+    assert!(text.bytes().all(|b| b.is_ascii_hexdigit()));
+    assert!(!shown(&made).contains(&text));
+
+    // A second init keeps the secret the first one made, since handles depend on it.
+    std::fs::remove_file(scratch.path("connector.cc")).unwrap();
+    let again = init(&scratch, &init_flags(&scratch, &[]));
+    assert!(again.status.success(), "{}", shown(&again));
+    assert_eq!(std::fs::read_to_string(&secret).unwrap(), text);
+
+    // A refused document leaves no secret file behind.
+    let fresh = Scratch::new();
+    let flags = init_flags(&fresh, &["--max-payload-bytes", "1"]);
+    let refused = init(&fresh, &flags);
+    assert!(!refused.status.success());
+    assert!(!fresh.path("handle.secret").exists());
+}
+
+#[cfg(unix)]
+#[test]
+fn check_wants_the_handle_secret_credential_and_names_it() {
+    let scratch = Scratch::new();
+    let (flags, credentials) = prepare(&scratch);
+    let made = init(&scratch, &flags);
+    assert!(made.status.success(), "{}", shown(&made));
+    let config = scratch.path("connector.cc");
+    let path = config.to_str().unwrap();
+    for body in [None, Some("too-short")] {
+        let secret = credentials.join("claude-handle-secret");
+        match body {
+            None => std::fs::remove_file(&secret).unwrap(),
+            Some(text) => std::fs::write(&secret, text).unwrap(),
+        }
+        let checked = run(
+            &["config", "check", path],
+            &[("CREDENTIALS_DIRECTORY", &credentials)],
+        );
+        assert!(!checked.status.success());
+        let text = shown(&checked);
+        assert!(text.contains("handle_secret_credential"), "{text}");
+        assert!(!text.contains("too-short"), "{text}");
+    }
+}
+
+/// A document written before the handle secret had a field cannot name where
+/// the secret is, so it is not a document this connector serves from: it is
+/// refused as a store fault, to be authored again with `config init`.
+#[test]
+fn a_document_without_the_handle_secret_field_is_refused_as_a_store() {
+    let scratch = Scratch::new();
+    let path = scratch.path("old.cc");
+    let mut payload = rmp_serde::to_vec(&valid()).unwrap();
+    // array16(23) -> array16(22), and drop the trailing 20-byte fixstr.
+    assert_eq!(payload[..3], [0xdc, 0, 23]);
+    payload[2] = 22;
+    payload.truncate(payload.len() - 21);
+    store_raw(
+        &path,
+        "runtime",
+        ClaudeConnectorConfig::TYPE,
+        Some(ClaudeConnectorConfig::TYPE),
+        payload,
+    );
+    assert!(matches!(load_config(&path), Err(ConfigError::Store)));
+}
+
+#[test]
+fn the_handle_secret_credential_is_a_name_of_its_own() {
+    for name in ["claude-token", "a/b", "", ".."] {
+        let mut config = valid();
+        config.handle_secret_credential = name.to_string();
+        assert_eq!(refused(&config), "handle_secret_credential", "{name:?}");
+    }
+}
+
+#[cfg(unix)]
+#[test]
+fn init_refuses_a_secret_file_it_cannot_create_and_writes_no_document() {
+    let scratch = Scratch::new();
+    let mut flags = init_flags(&scratch, &[]);
+    let at = flags
+        .iter()
+        .position(|f| f == "--handle-secret-file")
+        .unwrap();
+    flags[at + 1] = scratch
+        .path("absent/handle.secret")
+        .to_string_lossy()
+        .into_owned();
+    let made = init(&scratch, &flags);
+    assert!(!made.status.success());
+    assert!(!scratch.path("connector.cc").exists());
+}
+
+#[cfg(unix)]
+#[test]
+fn config_init_keeps_an_existing_secret_file_only_if_it_is_usable() {
+    use std::os::unix::fs::{PermissionsExt, symlink};
+
+    let write = |path: &Path, bytes: usize, mode: u32| {
+        std::fs::write(path, "k".repeat(bytes)).unwrap();
+        std::fs::set_permissions(path, std::fs::Permissions::from_mode(mode)).unwrap();
+    };
+    type Plant = fn(&Scratch, &Path);
+    let refused: Vec<(&str, Plant)> = vec![
+        ("world readable", |_, p| {
+            std::fs::write(p, "k".repeat(64)).unwrap();
+            std::fs::set_permissions(p, std::fs::Permissions::from_mode(0o644)).unwrap();
+        }),
+        ("group writable", |_, p| {
+            std::fs::write(p, "k".repeat(64)).unwrap();
+            std::fs::set_permissions(p, std::fs::Permissions::from_mode(0o660)).unwrap();
+        }),
+        ("read only", |_, p| {
+            std::fs::write(p, "k".repeat(64)).unwrap();
+            std::fs::set_permissions(p, std::fs::Permissions::from_mode(0o400)).unwrap();
+        }),
+        ("one byte short", |_, p| {
+            std::fs::write(p, "k".repeat(31)).unwrap();
+            std::fs::set_permissions(p, std::fs::Permissions::from_mode(0o600)).unwrap();
+        }),
+        ("empty", |_, p| {
+            std::fs::write(p, "").unwrap();
+            std::fs::set_permissions(p, std::fs::Permissions::from_mode(0o600)).unwrap();
+        }),
+        ("a directory", |_, p| {
+            std::fs::create_dir(p).unwrap();
+            std::fs::set_permissions(p, std::fs::Permissions::from_mode(0o600)).unwrap();
+        }),
+        ("a link to a good file", |s, p| {
+            let target = s.path("elsewhere.secret");
+            std::fs::write(&target, "k".repeat(64)).unwrap();
+            std::fs::set_permissions(&target, std::fs::Permissions::from_mode(0o600)).unwrap();
+            symlink(&target, p).unwrap();
+        }),
+    ];
+    for (name, plant) in refused {
+        let scratch = Scratch::new();
+        let flags = init_flags(&scratch, &[]);
+        let secret = scratch.path("handle.secret");
+        plant(&scratch, &secret);
+        let before = std::fs::symlink_metadata(&secret).unwrap();
+        let made = init(&scratch, &flags);
+        assert!(!made.status.success(), "{name} was accepted");
+        assert!(shown(&made).contains("--handle-secret-file"), "{name}");
+        assert!(!shown(&made).contains("kkkk"), "{name}");
+        assert!(
+            !scratch.path("connector.cc").exists(),
+            "{name} left a document"
+        );
+        let after = std::fs::symlink_metadata(&secret).unwrap();
+        assert_eq!(
+            before.permissions().mode(),
+            after.permissions().mode(),
+            "{name}"
+        );
+        assert_eq!(before.len(), after.len(), "{name}");
+        assert_eq!(
+            before.file_type().is_symlink(),
+            after.file_type().is_symlink(),
+            "{name}"
+        );
+    }
+
+    // The smallest usable file is accepted and left exactly as it was.
+    let scratch = Scratch::new();
+    let flags = init_flags(&scratch, &[]);
+    let secret = scratch.path("handle.secret");
+    write(&secret, 32, 0o600);
+    let made = init(&scratch, &flags);
+    assert!(made.status.success(), "{}", shown(&made));
+    assert_eq!(std::fs::read_to_string(&secret).unwrap(), "k".repeat(32));
+    assert!(scratch.path("connector.cc").exists());
+}
diff --git a/crates/model-connector/src/serve.rs b/crates/model-connector/src/serve.rs
index cf76afc..cd3663f 100644
--- a/crates/model-connector/src/serve.rs
+++ b/crates/model-connector/src/serve.rs
@@ -19,7 +19,7 @@ use crate::envelope::{
 use crate::readout::PlanQuotaSnapshot;
 use crate::service::{
     CallerAdmission, PriceTable, ProviderCall, SpendCeiling, SpendWindowMs, TransportAdmission,
-    TransportService,
+    TransportService, distinct_admissions,
 };
 
 /// The only provider seam of the connector kit. A connector binary implements
@@ -87,6 +87,7 @@ impl ServeLimits {
         if admissions.is_empty() {
             return Err(ServeError::InvalidConfig("callers"));
         }
+        distinct_admissions(admissions)?;
         let slots = admissions
             .iter()
             .try_fold(0_usize, |total, caller| {
@@ -950,6 +951,26 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn the_limits_refuse_callers_that_share_a_key_or_an_id() {
+        let limits = limits();
+        let shared_key = [
+            caller("one", "shared-key", 1),
+            caller("two", "shared-key", 1),
+        ];
+        assert!(matches!(
+            limits.validate(&shared_key),
+            Err(ServeError::Service(ServiceError::SharedCallerKey))
+        ));
+        let shared_id = [caller("one", "key-a", 1), caller("one", "key-b", 1)];
+        assert!(matches!(
+            limits.validate(&shared_id),
+            Err(ServeError::Service(ServiceError::DuplicateCaller))
+        ));
+        let distinct = [caller("one", "key-a", 1), caller("two", "key-b", 1)];
+        assert!(limits.validate(&distinct).is_ok());
+    }
+
     #[test]
     fn only_loopback_peers_are_admitted() {
         assert!(peer_admitted(IpAddr::from([127, 0, 0, 1])));
diff --git a/crates/model-connector/src/service.rs b/crates/model-connector/src/service.rs
index 16853dc..8adaf07 100644
--- a/crates/model-connector/src/service.rs
+++ b/crates/model-connector/src/service.rs
@@ -47,6 +47,24 @@ pub struct SpendCeiling(pub u64);
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub struct SpendWindowMs(pub u64);
 
+/// The rule that admitted callers are distinct: no two share a connection key
+/// (a key names its caller), and no two share a caller id. `TransportService`
+/// opens only under it, and a readiness check that must agree with opening
+/// asks the same function.
+pub fn distinct_admissions(admissions: &[CallerAdmission]) -> Result<(), ServiceError> {
+    let mut keys = HashSet::new();
+    let mut ids = HashSet::new();
+    for admission in admissions {
+        if !keys.insert(admission.security.0) {
+            return Err(ServiceError::SharedCallerKey);
+        }
+        if !ids.insert(admission.caller_runtime_id.as_str()) {
+            return Err(ServiceError::DuplicateCaller);
+        }
+    }
+    Ok(())
+}
+
 pub struct CallerAdmission {
     caller_runtime_id: String,
     connection_key_epoch: u32,
@@ -388,6 +406,12 @@ impl ProviderCall {
         &self.request
     }
 
+    /// The caller this call was admitted for: the caller whose key sealed the
+    /// invocation, never a name the request carries.
+    pub fn caller_runtime_id(&self) -> &str {
+        &self.receipt.caller_runtime_id
+    }
+
     /// How many foreign reasoning items the service removed from the request.
     pub fn foreign_reasoning_dropped(&self) -> u32 {
         self.receipt.foreign_reasoning_dropped
@@ -533,17 +557,12 @@ impl TransportService {
         {
             return Err(ServiceError::InvalidAdmission);
         }
-        let mut callers = HashMap::new();
-        let mut keys = HashSet::new();
-        for admission in admissions {
-            if !keys.insert(admission.security.0) {
-                return Err(ServiceError::SharedCallerKey);
-            }
-            let caller_runtime_id = admission.caller_runtime_id.clone();
-            if callers.insert(caller_runtime_id, admission).is_some() {
-                return Err(ServiceError::DuplicateCaller);
-            }
-        }
+        let admissions: Vec<CallerAdmission> = admissions.into_iter().collect();
+        distinct_admissions(&admissions)?;
+        let callers: HashMap<String, CallerAdmission> = admissions
+            .into_iter()
+            .map(|admission| (admission.caller_runtime_id.clone(), admission))
+            .collect();
         if callers.is_empty() {
             return Err(ServiceError::InvalidAdmission);
         }
@@ -3800,6 +3819,28 @@ mod tests {
         assert_eq!((idle.spent_micro_usd, idle.reserved_micro_usd), (0, 0));
     }
 
+    #[test]
+    fn any_admitted_caller_reads_every_callers_row() {
+        let store = TestStore::new();
+        let mut service = open_service(&store, 4);
+        let claim = claim_at(&mut service, "request-1", 1_000);
+        let result = text_result(claim.call(), "done");
+        service.complete(*claim, result, FRAME).unwrap();
+        // Ghostlight asks, with its own key, and sees epiphany's spend.
+        let readout = readout_as(
+            &service,
+            "ghostlight-yggdrasil",
+            "ghostlight-distinct-test-key",
+            1_500,
+            FRAME,
+        )
+        .unwrap();
+        assert_eq!(row(&readout, CALLER).spent_micro_usd, FIXTURE_USAGE_COST);
+        assert_eq!(row(&readout, "ghostlight-yggdrasil").spent_micro_usd, 0);
+        // A key that belongs to no admitted caller reads nothing.
+        assert!(readout_as(&service, "ghostlight-yggdrasil", KEY, 1_500, FRAME).is_err());
+    }
+
     #[test]
     fn a_claim_a_crash_left_reads_as_spent_not_reserved() {
         let store = TestStore::new();