feat: add authorized account lookup without provisioning

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a0e747-8f27-7242-8df8-8bc44f88c929

src/user_engine/service.py

@@ -2,6 +2,9 @@
 
 from __future__ import annotations
 
+import hashlib
+import json
+
 from dataclasses import dataclass, field, replace
 from datetime import datetime, timedelta
 from typing import Any, Callable, Iterable, Mapping
@@ -105,6 +108,21 @@ class ReadinessReport:
     checks: Mapping[str, bool]
 
 
+@dataclass(frozen=True)
+class AccountAuthority:
+    """Current account state only; not a root entitlement or tenant lifecycle grant."""
+    issuer: str
+    subject: str
+    tenant: str
+    user_id: str
+    account_id: str
+    identity_id: str
+    account_status: AccountStatus
+    tenant_account_status: AccountStatus | None
+    observed_at: datetime
+    decision_id: str | None
+
+
 @dataclass(frozen=True)
 class UserSession:
     actor: Actor
@@ -1766,6 +1784,44 @@ class UserEngineService:
             )
         return self._session(actor, user, account)
 
+    def lookup_account_authority(
+        self, caller: Actor, *, issuer: str, subject: str, tenant: str,
+        correlation_id: str | None = None,
+    ) -> AccountAuthority:
+        """Authorize a read of an existing identity; never call provisioning /me.
+
+        The host must authenticate caller independently of the lookup target.
+        This action needs its own policy grant; caller roles do not grant it here.
+        Missing scoped accounts are explicit, never defaulted to active.
+        """
+        if any(not isinstance(value, str) or not value.strip()
+               for value in (issuer, subject, tenant)):
+            raise ValidationError("exact issuer, subject and tenant are required")
+        correlation_id = correlation_id or new_id("corr")
+        lookup_id = "sha256:" + hashlib.sha256(json.dumps(
+            [issuer, subject, tenant], separators=(",", ":"),
+        ).encode()).hexdigest()
+        decision = self._authorize(
+            caller, action="account.authority.read",
+            resource_type="user-engine:account-authority", resource_id=lookup_id,
+            tenant=tenant, correlation_id=correlation_id,
+            context={"lookup_issuer": issuer, "lookup_subject": subject},
+        )
+        observed_at = utc_now()
+        identity = self.store.find_identity(issuer, subject)
+        if identity is None:
+            raise NotFoundError("identity link not found")
+        user = self._require_user(identity.user_id)
+        account = self._require_account(user.user_id)
+        scoped = self.store.tenant_account(tenant, user.user_id)
+        return AccountAuthority(
+            issuer=issuer, subject=subject, tenant=tenant, user_id=user.user_id,
+            account_id=account.account_id, identity_id=identity.identity_id,
+            account_status=account.status,
+            tenant_account_status=scoped.status if scoped is not None else None,
+            observed_at=observed_at, decision_id=decision.decision_id,
+        )
+
     def create_user(
         self,
         actor: Actor,

tests/test_account_authority.py

@@ -0,0 +1,69 @@
+import unittest
+from dataclasses import replace
+
+from user_engine.adapters.local import InMemoryUserEngineStore
+from user_engine.domain import AccountStatus, AuthorizationEffect
+from user_engine.errors import AuthorizationDenied, NotFoundError
+from user_engine.service import UserEngineService
+from user_engine.testing.fixtures import FixtureIdentityClaimsAdapter, StaticAuthorizationCheckPort, human_actor_claims
+
+
+class AccountAuthorityTests(unittest.TestCase):
+    def setUp(self):
+        self.store = InMemoryUserEngineStore()
+        self.policy = StaticAuthorizationCheckPort()
+        self.service = UserEngineService(store=self.store,identity_adapter=FixtureIdentityClaimsAdapter(),authorization=self.policy)
+        self.session = self.service.me(human_actor_claims())
+        self.caller = replace(self.session.actor,subject='fixture-workload')
+
+    def lookup(self,**overrides):
+        return self.service.lookup_account_authority(self.caller,**{
+            'issuer':self.session.actor.issuer,'subject':self.session.actor.subject,
+            'tenant':self.session.actor.tenant,**overrides})
+
+    def test_existing_lookup_does_not_mutate_and_keeps_caller_distinct(self):
+        before = repr(self.store.__dict__)
+        result = self.lookup()
+        self.assertEqual(repr(self.store.__dict__),before)
+        self.assertEqual(result.identity_id,self.session.identities[0].identity_id)
+        self.assertEqual(result.account_status,AccountStatus.ACTIVE)
+        self.assertEqual(result.tenant_account_status,AccountStatus.ACTIVE)
+        request = self.policy.requests[-1]
+        self.assertEqual(request.action,'account.authority.read')
+        self.assertEqual(request.actor.subject,'fixture-workload')
+        self.assertEqual(request.context['lookup_subject'],self.session.actor.subject)
+        self.assertIsNotNone(result.observed_at.tzinfo)
+
+    def test_unknown_identity_does_not_provision(self):
+        before = repr(self.store.__dict__)
+        with self.assertRaises(NotFoundError): self.lookup(subject='unknown')
+        with self.assertRaises(NotFoundError): self.lookup(issuer='https://other.example')
+        self.assertEqual(repr(self.store.__dict__),before)
+
+    def test_missing_scoped_account_is_explicit(self):
+        self.assertIsNone(self.lookup(tenant='unknown-tenant').tenant_account_status)
+
+    def test_current_global_and_scoped_status_are_returned_each_time(self):
+        self.lookup()
+        self.store.save_account(replace(self.session.account,status=AccountStatus.DISABLED))
+        scoped = self.store.tenant_account(self.session.actor.tenant,self.session.user.user_id)
+        self.store.save_tenant_account(replace(scoped,status=AccountStatus.DISABLED))
+        result = self.lookup()
+        self.assertEqual(result.account_status,AccountStatus.DISABLED)
+        self.assertEqual(result.tenant_account_status,AccountStatus.DISABLED)
+
+    def test_denial_precedes_target_identity_lookup(self):
+        self.policy.effect = AuthorizationEffect.DENY
+        original = self.store.find_identity
+        def checked(issuer,subject):
+            if subject == self.session.actor.subject:
+                raise AssertionError('target lookup before authorization')
+            return original(issuer,subject)
+        self.store.find_identity = checked
+        with self.assertRaises(AuthorizationDenied): self.lookup()
+
+    def test_does_not_emit_root_grants_or_profile_details(self):
+        result = self.lookup()
+        self.assertFalse(hasattr(result,'root_entitled'))
+        self.assertFalse(hasattr(result,'roles'))
+        self.assertFalse(hasattr(result,'primary_email'))

workplans/USER-WP-0037-hub-authority-lookup.md

@@ -0,0 +1,70 @@
+---
+id: USER-WP-0037
+type: workplan
+title: "Read-only account authority for Hub integration"
+domain: communication
+repo: user-engine
+status: active
+owner: codex
+topic_slug: communication
+created: "2026-09-28"
+updated: "2026-09-28"
+related:
+  - HUB-WP-0012
+---
+
+# Hub authority lookup
+
+Owner-side implementation for HUB-WP-0012. Source work is authorized; real
+workload grants, root entitlements and deployment acceptance remain separate.
+
+## Existing account lookup without provisioning
+
+```task
+id: USER-WP-0037-T01
+status: done
+priority: high
+```
+
+Add an authorized service lookup for exact issuer/subject and scoped tenant.
+Authorize the caller separately before target lookup. Unknown identities never
+provision; return current global/scoped account state without roles or profile
+PII. Preserve source observation time and identity/account references.
+
+Implemented `UserEngineService.lookup_account_authority` with the separate
+`account.authority.read` action and a digest-bound issuer/subject/tenant resource.
+Caller identity is distinct from the target. Missing tenant-account state is
+explicit `None`, not active. AccountAuthority contains no root grant, token roles
+or profile/email data. Authorization denial may emit the existing audit record;
+this operation does not provision or mutate domain records.
+
+Validation: `make test` ran 270 tests successfully (eight optional skips), including
+six new lookup checks; layer conformance passed. No HTTP endpoint, live grant or
+production deployment was added. T02/T03 retain those outstanding owner gates.
+
+## Root entitlement and mapping disposition
+
+```task
+id: USER-WP-0037-T02
+status: todo
+priority: high
+```
+
+With NetKingdom, specify explicit current root-grant custody/revocation and
+platform:root to tenant:platform mapping. Do not derive a grant from JWT roles,
+a username, or Hub's configured root subject. Define retainable evidence for
+account and entitlement observations.
+
+## Authenticated workload transport and private acceptance
+
+```task
+id: USER-WP-0037-T03
+status: todo
+priority: high
+```
+
+Expose a narrowly scoped read API with an independently authenticated Hub
+workload, policy action account.authority.read, exact target binding and no
+end-user bearer forwarding. Add HTTP schema and denial/no-provisioning tests,
+then admit real caller credentials through their owner and verify withdrawal,
+outages and private Hub integration. A service method alone is not this gate.