feat: derive agent identity from a single private-key env var

docs/site/reference/environment-variables.mdx

@@ -11,8 +11,25 @@ Authsome interacts with environment variables in three roles: **inputs** that ch
 |----------|---------|
 | `AUTHSOME_BASE_URL` | The daemon URL. On client machines, set this to point the CLI and proxy at a remote daemon instead of auto-starting one locally. On the daemon host, set this to the public URL when behind a reverse proxy — authsome uses it to build OAuth callback URLs such as `/auth/callback/oauth`. Defaults to `http://127.0.0.1:7998`. |
 | `AUTHSOME_HOME` | Override the default `~/.authsome` directory. Useful for tests, ephemeral environments, and per-project vaults. |
+| `AUTHSOME_IDENTITY_PRIVATE_KEY` | Hex-encoded Ed25519 private key. The agent's full identity is derived from this single value: the DID is computed locally and the human-readable handle is resolved from the identity server. Sufficient on its own — no `authsome init` or local identity files required. |
+| `AUTHSOME_IDENTITY` | *(Optional, deprecated)* Explicit handle override used alongside `AUTHSOME_IDENTITY_PRIVATE_KEY`. No longer required: the handle is resolved from the identity server. See the migration note below. |
 | `HTTP_PROXY` / `HTTPS_PROXY` | Honored by authsome's own outbound HTTP requests (token endpoints, device flow polling). The proxy started by `authsome run` is **set** as these variables in the child process; it does not chain through them. |
 
+### Agent identity from a single key
+
+For headless, CI, and container deployments, supply only the private key:
+
+```bash
+export AUTHSOME_IDENTITY_PRIVATE_KEY=<hex-encoded-ed25519-key>
+uv run authsome list
+```
+
+On startup the CLI derives the DID from the key and asks the identity server for the handle bound to that DID. If the DID is not yet registered, a handle is generated and registered automatically, then the standard browser claim flow runs.
+
+<Note>
+**Migration:** the previous setup required both `AUTHSOME_IDENTITY` (handle) and `AUTHSOME_IDENTITY_PRIVATE_KEY`. The key alone is now sufficient. `AUTHSOME_IDENTITY` is still honored as an explicit handle override but is deprecated and no longer required. Resolving the handle requires the daemon to be reachable at CLI startup.
+</Note>
+
 ### `AUTHSOME_BASE_URL` for remote daemons
 
 For remote daemon deployments:

src/authsome/cli/client.py

@@ -16,6 +16,7 @@
 import authsome.errors as err_mod
 from authsome.cli.identity import RuntimeIdentity
 from authsome.config import get_authsome_config
+from authsome.identity.helpers import generate_handle
 from authsome.identity.proof import POP_AUTH_SCHEME, create_proof_jwt
 from authsome.server.config import get_server_config
 
@@ -130,6 +131,8 @@ async def _request(
 
     async def _proof_headers(self, method: str, path: str, body: bytes) -> dict[str, str]:
         identity = await self.ensure_identity_ready()
+        if identity.handle is None:
+            raise RuntimeError("Identity handle could not be resolved from the identity server")
         token = create_proof_jwt(
             private_key=identity.signer,
             issuer=identity.did,
@@ -150,7 +153,11 @@ async def ensure_identity_ready(self) -> RuntimeIdentity:
         runtime = self._runtime_identity()
         if self._server_registered:
             return runtime
-        await self._check_server_registration(runtime)
+        if runtime.handle is None:
+            runtime = await self._resolve_env_identity(runtime)
+            self._identity = runtime
+        runtime = await self._check_server_registration(runtime)
+        self._identity = runtime
         self._server_registered = True
         return runtime
 
@@ -159,23 +166,49 @@ def _runtime_identity(self) -> RuntimeIdentity:
             self._identity = RuntimeIdentity.load(self._home)
         return self._identity
 
-    async def _check_server_registration(self, runtime: RuntimeIdentity) -> None:
+    async def _resolve_env_identity(self, runtime: RuntimeIdentity) -> RuntimeIdentity:
+        """Resolve a handle-less env identity's handle from the identity server.
+
+        Looks up the handle bound to the DID. When the DID is unknown the agent
+        is brand new, so a handle is generated; the existing registration/claim
+        flow then registers it.
+        """
+        handle = await self.resolve_handle_by_did(runtime.did)
+        if handle is None:
+            handle = generate_handle()
+        return runtime.model_copy(update={"handle": handle})
+
+    async def _check_server_registration(self, runtime: RuntimeIdentity) -> RuntimeIdentity:
         """Verify registration with the server; register and claim if needed."""
+        handle = runtime.handle
+        if handle is None:
+            raise RuntimeError("Identity handle could not be resolved from the identity server")
         try:
-            identity_status = await self.get_identity_status(runtime.handle)
+            identity_status = await self.get_identity_status(handle)
         except httpx.HTTPStatusError as exc:
             if exc.response.status_code != status.HTTP_404_NOT_FOUND:
                 raise
-            identity_status = await self.register_identity(runtime.handle, runtime.did)
+            try:
+                identity_status = await self.register_identity(handle, runtime.did)
+            except httpx.HTTPStatusError as reg_exc:
+                if reg_exc.response.status_code != status.HTTP_409_CONFLICT:
+                    raise
+                resolved_handle = await self.resolve_handle_by_did(runtime.did)
+                if resolved_handle is None:
+                    raise
+                handle = resolved_handle
+                runtime = runtime.model_copy(update={"handle": handle})
+                identity_status = await self.get_identity_status(handle)
 
         reg_status = identity_status.get("registration_status", "")
         if reg_status == "claim_required":
             claim_url = identity_status.get("claim_url", "")
             if claim_url:
                 self._open_claim_url(claim_url)
-            await self._poll_claim_completion(runtime.handle)
+            await self._poll_claim_completion(handle)
         elif reg_status == "rejected":
-            raise RuntimeError(f"Agent '{runtime.handle}' claim was rejected by the server")
+            raise RuntimeError(f"Agent '{handle}' claim was rejected by the server")
+        return runtime
 
     def _open_claim_url(self, claim_url: str) -> None:
         print(f"Open this URL in your browser to claim this agent:\n  {claim_url}", file=sys.stderr)
@@ -259,6 +292,16 @@ async def register_identity(self, handle: str, did: str) -> dict[str, Any]:
     async def get_identity_status(self, handle: str) -> dict[str, Any]:
         return await self._get(f"{API_PREFIX}/identities/{handle}", protected=False)
 
+    async def resolve_handle_by_did(self, did: str) -> str | None:
+        """Return the handle the identity server has bound to ``did``, or None if unknown."""
+        try:
+            payload = await self._get(f"{API_PREFIX}/identities/by-did/{did}", protected=False)
+        except httpx.HTTPStatusError as exc:
+            if exc.response.status_code == status.HTTP_404_NOT_FOUND:
+                return None
+            raise
+        return payload.get("identity")
+
     async def remove(self, provider: str) -> None:
         await self._delete(f"{API_PREFIX}/providers/{provider}")
 

src/authsome/cli/identity.py

@@ -23,9 +23,14 @@
 
 
 class RuntimeIdentity(BaseModel):
-    """Resolved acting identity for the current process."""
+    """Resolved acting identity for the current process.
 
-    handle: str
+    ``handle`` is server-registered metadata, not part of the cryptographic
+    identity. When only a private key is supplied via the environment the handle
+    is left unresolved (``None``) and filled in later from the identity server.
+    """
+
+    handle: str | None = None
     did: str
     signer: Ed25519PrivateKey
 
@@ -35,21 +40,32 @@ class RuntimeIdentity(BaseModel):
     def from_pkey(cls, handle: str, signer: Ed25519PrivateKey) -> Self:
         return cls(handle=validate_handle(handle), did=public_key_to_did_key(signer.public_key()), signer=signer)
 
+    @classmethod
+    def from_env_private_key(cls, signer: Ed25519PrivateKey) -> Self:
+        """Build a handle-less identity from a private key; the handle is resolved later."""
+        return cls(handle=None, did=public_key_to_did_key(signer.public_key()), signer=signer)
+
     @classmethod
     def from_filesystem(cls, home: Path, handle: str) -> Self:
         metadata = cls.load_metadata(home, handle)
         return cls(handle=metadata.handle, did=metadata.did, signer=cls.load_private_key(home, handle))
 
     @classmethod
     def load(cls, home: Path, env: Mapping[str, str] | None = None) -> Self:
-        """Resolve the acting process identity from env or local identity files."""
+        """Resolve the acting process identity from env or local identity files.
+
+        A private key alone is sufficient: the DID is derived locally and the
+        handle is resolved from the identity server before first use. Supplying
+        ``AUTHSOME_IDENTITY`` alongside the key keeps the explicit handle.
+        """
         handle_override, private_key_hex = cls._env_identity_values(env)
-        if private_key_hex and not handle_override:
-            raise ValueError("AUTHSOME_IDENTITY_PRIVATE_KEY requires AUTHSOME_IDENTITY")
 
         if handle_override and private_key_hex:
             return cls.from_pkey(handle_override, private_key_from_hex(private_key_hex))
 
+        if private_key_hex:
+            return cls.from_env_private_key(private_key_from_hex(private_key_hex))
+
         return cls.ensure_local(home, active_handle=handle_override)
 
     @classmethod

src/authsome/server/routes/identities.py

@@ -55,6 +55,14 @@ async def register_identity(body: RegisterIdentityRequest, request: Request) ->
     return registration_status.to_payload()
 
 
+@router.get("/by-did/{did:path}")
+async def resolve_identity_by_did(did: str, request: Request) -> dict[str, str]:
+    registration = await request.app.state.store.identity_registry.resolve_by_did(did)
+    if registration is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Identity not found")
+    return {"identity": registration.handle, "did": registration.did}
+
+
 @router.get("/{handle}")
 async def get_identity_status(handle: str, request: Request) -> dict[str, str]:
     registration_status = await request.app.state.identity_bootstrap.get_identity_status(handle=handle)

src/authsome/server/store/repositories.py

@@ -428,6 +428,14 @@ async def register(self, *, handle: str, did: str) -> IdentityRegistration:
 
     async def resolve(self, handle: str) -> IdentityRegistration | None:
         row = await self._db.fetch_one("SELECT * FROM identity_registrations WHERE handle = ?", [handle])
+        return self._registration_from_row(row)
+
+    async def resolve_by_did(self, did: str) -> IdentityRegistration | None:
+        row = await self._db.fetch_one("SELECT * FROM identity_registrations WHERE did = ?", [did])
+        return self._registration_from_row(row)
+
+    @staticmethod
+    def _registration_from_row(row: Any | None) -> IdentityRegistration | None:
         if row is None:
             return None
         return IdentityRegistration(

tests/cli/test_client_signing.py

@@ -311,14 +311,140 @@ def fake_request(method, url, data=None, headers=None, timeout=None):
 
 
 @pytest.mark.asyncio
-async def test_env_identity_private_key_without_handle_errors(monkeypatch, tmp_path: Path) -> None:
+async def test_env_private_key_only_resolves_handle_from_server(monkeypatch, tmp_path: Path) -> None:
     monkeypatch.setenv("AUTHSOME_HOME", str(tmp_path))
-    monkeypatch.setenv("AUTHSOME_IDENTITY_PRIVATE_KEY", "00" * 32)
+    source = RuntimeIdentity.create(tmp_path, "steady-wisely-boldly-0042")
+    private_key_hex = RuntimeIdentity.key_path(tmp_path, source.handle).read_text(encoding="utf-8").strip()
+    RuntimeIdentity.key_path(tmp_path, source.handle).unlink()
+    RuntimeIdentity.metadata_path(tmp_path, source.handle).unlink()
+    monkeypatch.setenv("AUTHSOME_IDENTITY_PRIVATE_KEY", private_key_hex)
+    resolved_handle = "resolved-from-server-0001"
+    calls: list[tuple[str, str]] = []
 
-    client = AuthsomeApiClient("http://127.0.0.1:7998")
+    def fake_request(method, url, data=None, headers=None, timeout=None):
+        calls.append((method, url))
+        response = Mock()
+        response.status_code = status.HTTP_200_OK
+        response.raise_for_status.return_value = None
+        if "/api/identities/by-did/" in url:
+            response.json.return_value = {"identity": resolved_handle, "did": source.did}
+        elif f"/api/identities/{resolved_handle}" in url:
+            response.json.return_value = {"identity": resolved_handle, "registration_status": "claimed"}
+        else:
+            response.json.return_value = {"connections": [], "by_source": {"bundled": [], "custom": []}}
+        return response
+
+    _patch_httpx_request(monkeypatch, fake_request)
+
+    identity = await AuthsomeApiClient("http://127.0.0.1:7998").ensure_identity_ready()
+
+    assert identity.handle == resolved_handle
+    assert identity.did == source.did
+    assert (
+        "GET",
+        f"http://127.0.0.1:7998/api/identities/by-did/{source.did}",
+    ) in calls
+
+
+@pytest.mark.asyncio
+async def test_env_private_key_only_registers_generated_handle_when_did_unknown(monkeypatch, tmp_path: Path) -> None:
+    monkeypatch.setenv("AUTHSOME_HOME", str(tmp_path))
+    source = RuntimeIdentity.create(tmp_path, "steady-wisely-boldly-0042")
+    private_key_hex = RuntimeIdentity.key_path(tmp_path, source.handle).read_text(encoding="utf-8").strip()
+    RuntimeIdentity.key_path(tmp_path, source.handle).unlink()
+    RuntimeIdentity.metadata_path(tmp_path, source.handle).unlink()
+    monkeypatch.setenv("AUTHSOME_IDENTITY_PRIVATE_KEY", private_key_hex)
+    registered: dict = {}
+
+    def fake_request(method, url, data=None, headers=None, timeout=None):
+        response = Mock()
+        if "/api/identities/by-did/" in url:
+            response.status_code = status.HTTP_404_NOT_FOUND
+            response.raise_for_status.side_effect = httpx.HTTPStatusError(
+                "Not Found", request=Mock(), response=Mock(status_code=status.HTTP_404_NOT_FOUND)
+            )
+        elif url.endswith("/api/identities/register"):
+            response.status_code = status.HTTP_200_OK
+            response.raise_for_status.return_value = None
+            registered.update(json.loads(data.decode("utf-8")))
+            response.json.return_value = {
+                "identity": registered["handle"],
+                "did": registered["did"],
+                "registration_status": "claimed",
+            }
+        elif "/api/identities/" in url and method == "GET":
+            response.status_code = status.HTTP_404_NOT_FOUND
+            response.raise_for_status.side_effect = httpx.HTTPStatusError(
+                "Not Found", request=Mock(), response=Mock(status_code=status.HTTP_404_NOT_FOUND)
+            )
+        else:
+            response.status_code = status.HTTP_200_OK
+            response.raise_for_status.return_value = None
+            response.json.return_value = {"connections": [], "by_source": {"bundled": [], "custom": []}}
+        return response
+
+    _patch_httpx_request(monkeypatch, fake_request)
+
+    identity = await AuthsomeApiClient("http://127.0.0.1:7998").ensure_identity_ready()
+
+    assert identity.handle is not None
+    assert registered["handle"] == identity.handle
+    assert registered["did"] == source.did
+
+
+@pytest.mark.asyncio
+async def test_env_private_key_only_recovers_when_register_races_on_did(monkeypatch, tmp_path: Path) -> None:
+    monkeypatch.setenv("AUTHSOME_HOME", str(tmp_path))
+    source = RuntimeIdentity.create(tmp_path, "steady-wisely-boldly-0042")
+    private_key_hex = RuntimeIdentity.key_path(tmp_path, source.handle).read_text(encoding="utf-8").strip()
+    RuntimeIdentity.key_path(tmp_path, source.handle).unlink()
+    RuntimeIdentity.metadata_path(tmp_path, source.handle).unlink()
+    monkeypatch.setenv("AUTHSOME_IDENTITY_PRIVATE_KEY", private_key_hex)
+    existing_handle = "winner-handle-0001"
+    by_did_calls: list[str] = []
+
+    def fake_request(method, url, data=None, headers=None, timeout=None):
+        response = Mock()
+        if "/api/identities/by-did/" in url:
+            by_did_calls.append(url)
+            if len(by_did_calls) == 1:
+                response.status_code = status.HTTP_404_NOT_FOUND
+                response.raise_for_status.side_effect = httpx.HTTPStatusError(
+                    "Not Found", request=Mock(), response=Mock(status_code=status.HTTP_404_NOT_FOUND)
+                )
+            else:
+                response.status_code = status.HTTP_200_OK
+                response.raise_for_status.return_value = None
+                response.json.return_value = {"identity": existing_handle, "did": source.did}
+        elif url.endswith("/api/identities/register"):
+            response.status_code = status.HTTP_409_CONFLICT
+            response.raise_for_status.side_effect = httpx.HTTPStatusError(
+                "Conflict",
+                request=Mock(),
+                response=Mock(status_code=status.HTTP_409_CONFLICT),
+            )
+        elif f"/api/identities/{existing_handle}" in url:
+            response.status_code = status.HTTP_200_OK
+            response.raise_for_status.return_value = None
+            response.json.return_value = {"identity": existing_handle, "registration_status": "claimed"}
+        elif "/api/identities/" in url and method == "GET":
+            response.status_code = status.HTTP_404_NOT_FOUND
+            response.raise_for_status.side_effect = httpx.HTTPStatusError(
+                "Not Found", request=Mock(), response=Mock(status_code=status.HTTP_404_NOT_FOUND)
+            )
+        else:
+            response.status_code = status.HTTP_200_OK
+            response.raise_for_status.return_value = None
+            response.json.return_value = {"connections": [], "by_source": {"bundled": [], "custom": []}}
+        return response
+
+    _patch_httpx_request(monkeypatch, fake_request)
+
+    identity = await AuthsomeApiClient("http://127.0.0.1:7998").ensure_identity_ready()
 
-    with pytest.raises(ValueError, match="AUTHSOME_IDENTITY"):
-        await client.ensure_identity_ready()
+    assert identity.handle == existing_handle
+    assert identity.did == source.did
+    assert len(by_did_calls) > 1
 
 
 @pytest.mark.asyncio

tests/identity/test_identity.py

@@ -95,6 +95,11 @@ def test_runtime_identity_creates_missing_handle_override(tmp_path: Path) -> Non
     assert RuntimeIdentity.key_path(tmp_path, runtime.handle).exists()
 
 
-def test_runtime_identity_rejects_private_key_without_handle(tmp_path: Path) -> None:
-    with pytest.raises(ValueError, match="AUTHSOME_IDENTITY"):
-        RuntimeIdentity.load(tmp_path, env={"AUTHSOME_IDENTITY_PRIVATE_KEY": "00" * 32})
+def test_runtime_identity_env_private_key_only_defers_handle(tmp_path: Path) -> None:
+    source = RuntimeIdentity.create(tmp_path, "steady-wisely-boldly-0042")
+    private_key_hex = RuntimeIdentity.key_path(tmp_path, source.handle).read_text(encoding="utf-8").strip()
+
+    runtime = RuntimeIdentity.load(tmp_path, env={"AUTHSOME_IDENTITY_PRIVATE_KEY": private_key_hex})
+
+    assert runtime.handle is None
+    assert runtime.did == source.did