diff --git a/changelog.d/authorization-session-consistency.fixed.md b/changelog.d/authorization-session-consistency.fixed.md
new file mode 100644
index 0000000000..49b214183e
--- /dev/null
+++ b/changelog.d/authorization-session-consistency.fixed.md
@@ -0,0 +1,3 @@
+- Align authorization and session lifecycle handling across API services and
+  browser clients. Add regression coverage for resource resolution, session
+  transitions, upload workflows, and local asset serving.
diff --git a/config/graphql/agent_types.py b/config/graphql/agent_types.py
index 5d1ac79872..3c0a51a3e4 100644
--- a/config/graphql/agent_types.py
+++ b/config/graphql/agent_types.py
@@ -78,17 +78,31 @@ def name(self, info: strawberry.Info) -> str:
     corpus: Annotated[CorpusType, strawberry.lazy("config.graphql.corpus_types")] = (
         strawberry.field(name="corpus", default=None)
     )
-    fieldset: None | (
-        Annotated[FieldsetType, strawberry.lazy("config.graphql.extract_types")]
-    ) = strawberry.field(name="fieldset", default=None)
-    analyzer: None | (
-        Annotated[AnalyzerType, strawberry.lazy("config.graphql.extract_types")]
-    ) = strawberry.field(name="analyzer", default=None)
-    agent_config: AgentConfigurationType | None = strawberry.field(
+
+    @strawberry.field(name="fieldset")
+    def fieldset(
+        self, info: strawberry.Info
+    ) -> (
+        None | Annotated[FieldsetType, strawberry.lazy("config.graphql.extract_types")]
+    ):
+        return resolve_visible_fk(self, info, "fieldset_id", "FieldsetType")
+
+    @strawberry.field(name="analyzer")
+    def analyzer(
+        self, info: strawberry.Info
+    ) -> (
+        None | Annotated[AnalyzerType, strawberry.lazy("config.graphql.extract_types")]
+    ):
+        return resolve_visible_fk(self, info, "analyzer_id", "AnalyzerType")
+
+    @strawberry.field(
         name="agentConfig",
         description="Optional agent configuration for persona/tool defaults. Not required for agent actions — task_instructions alone is sufficient.",
-        default=None,
     )
+    def agent_config(self, info: strawberry.Info) -> AgentConfigurationType | None:
+        return resolve_visible_fk(
+            self, info, "agent_config_id", "AgentConfigurationType"
+        )
 
     @strawberry.field(
         name="taskInstructions",
@@ -576,13 +590,18 @@ def conversation(
     ):
         return resolve_visible_fk(self, info, "conversation_id", "ConversationType")
 
-    message: None | (
-        Annotated[MessageType, strawberry.lazy("config.graphql.conversation_types")]
-    ) = strawberry.field(
+    @strawberry.field(
         name="message",
         description="The message that triggered this execution (for NEW_MESSAGE trigger)",
-        default=None,
     )
+    def message(
+        self, info: strawberry.Info
+    ) -> (
+        None
+        | Annotated[MessageType, strawberry.lazy("config.graphql.conversation_types")]
+    ):
+        return resolve_visible_fk(self, info, "message_id", "MessageType")
+
     corpus: Annotated[CorpusType, strawberry.lazy("config.graphql.corpus_types")] = (
         strawberry.field(
             name="corpus",
@@ -644,20 +663,24 @@ def affected_objects(self, info: strawberry.Info) -> list[JSONString | None] | N
         description="Detailed agent result (for agent actions only)",
         default=None,
     )
-    extract: None | (
-        Annotated[ExtractType, strawberry.lazy("config.graphql.extract_types")]
-    ) = strawberry.field(
-        name="extract",
-        description="Extract created (for fieldset actions only)",
-        default=None,
+
+    @strawberry.field(
+        name="extract", description="Extract created (for fieldset actions only)"
     )
-    analysis: None | (
-        Annotated[AnalysisType, strawberry.lazy("config.graphql.extract_types")]
-    ) = strawberry.field(
-        name="analysis",
-        description="Analysis created (for analyzer actions only)",
-        default=None,
+    def extract(
+        self, info: strawberry.Info
+    ) -> None | Annotated[ExtractType, strawberry.lazy("config.graphql.extract_types")]:
+        return resolve_visible_fk(self, info, "extract_id", "ExtractType")
+
+    @strawberry.field(
+        name="analysis", description="Analysis created (for analyzer actions only)"
     )
+    def analysis(
+        self, info: strawberry.Info
+    ) -> (
+        None | Annotated[AnalysisType, strawberry.lazy("config.graphql.extract_types")]
+    ):
+        return resolve_visible_fk(self, info, "analysis_id", "AnalysisType")
 
     @strawberry.field(
         name="errorMessage", description="Error message if status is FAILED"
@@ -957,13 +980,17 @@ def triggering_conversation(
             self, info, "triggering_conversation_id", "ConversationType"
         )
 
-    triggering_message: None | (
-        Annotated[MessageType, strawberry.lazy("config.graphql.conversation_types")]
-    ) = strawberry.field(
+    @strawberry.field(
         name="triggeringMessage",
         description="Message that triggered this agent action (for NEW_MESSAGE trigger)",
-        default=None,
     )
+    def triggering_message(
+        self, info: strawberry.Info
+    ) -> (
+        None
+        | Annotated[MessageType, strawberry.lazy("config.graphql.conversation_types")]
+    ):
+        return resolve_visible_fk(self, info, "triggering_message_id", "MessageType")
 
     @strawberry.field(name="status")
     def status(
@@ -1148,11 +1175,14 @@ def name(self, info: strawberry.Info) -> str:
     def description(self, info: strawberry.Info) -> str:
         return coerce_str(getattr(self, "description", None))
 
-    agent_config: AgentConfigurationType | None = strawberry.field(
+    @strawberry.field(
         name="agentConfig",
         description="Optional agent configuration for persona/tool defaults.",
-        default=None,
     )
+    def agent_config(self, info: strawberry.Info) -> AgentConfigurationType | None:
+        return resolve_visible_fk(
+            self, info, "agent_config_id", "AgentConfigurationType"
+        )
 
     @strawberry.field(name="preAuthorizedTools")
     def pre_authorized_tools(self, info: strawberry.Info) -> list[str | None] | None:
diff --git a/config/graphql/annotation_types.py b/config/graphql/annotation_types.py
index 2dd6a71067..c37bbf2b9c 100644
--- a/config/graphql/annotation_types.py
+++ b/config/graphql/annotation_types.py
@@ -73,6 +73,7 @@
     is_authority_admin,
 )
 from opencontractserver.shared.services.base import BaseService
+from opencontractserver.shared.services.tree_traversal import TreeTraversalService
 from opencontractserver.utils.permissioning import get_users_permissions_for_obj
 
 
@@ -210,110 +211,48 @@ def _resolve_AnnotationType_feedback_count(root, info):
 
 
 def _resolve_AnnotationType_all_source_node_in_relationship(root, info):
-    return root.source_node_in_relationships.all()
+    return BaseService.filter_visible(
+        Relationship, info.context.user, request=info.context
+    ).filter(source_annotations=root)
 
 
 def _resolve_AnnotationType_all_target_node_in_relationship(root, info):
-    return root.target_node_in_relationships.all()
+    return BaseService.filter_visible(
+        Relationship, info.context.user, request=info.context
+    ).filter(target_annotations=root)
 
 
 def _resolve_AnnotationType_descendants_tree(root, info):
-    """
-    Returns a flat list of descendant annotations,
-    each including only the IDs of its immediate children.
-    """
-    from django_cte import CTE, with_cte
-
-    def get_descendants(cte):
-        base_qs = Annotation.objects.filter(parent_id=root.id).values(
-            "id", "parent_id", "raw_text"
-        )
-        recursive_qs = cte.join(Annotation, parent_id=cte.col.id).values(
-            "id", "parent_id", "raw_text"
-        )
-        return base_qs.union(recursive_qs, all=True)
-
-    cte = CTE.recursive(get_descendants)
-    descendants_qs = with_cte(cte, select=cte.queryset()).order_by("id")
-    descendants_list = list(descendants_qs)
-
-    return build_flat_tree(
-        descendants_list, type_name="AnnotationType", text_key="raw_text"
+    nodes = TreeTraversalService.get_nodes(
+        root,
+        info.context.user,
+        mode="descendants",
+        text_field="raw_text",
+        request=info.context,
     )
+    return build_flat_tree(nodes, type_name="AnnotationType", text_key="raw_text")
 
 
 def _resolve_AnnotationType_full_tree(root, info):
-    """
-    Returns a flat list of annotations from the root ancestor,
-    each including only the IDs of its immediate children.
-    """
-    from django_cte import CTE, with_cte
-
-    # Find the root ancestor
-    tree_root = root
-    while tree_root.parent_id is not None:
-        tree_root = tree_root.parent
-
-    def get_full_tree(cte):
-        base_qs = Annotation.objects.filter(id=tree_root.id).values(
-            "id", "parent_id", "raw_text"
-        )
-        recursive_qs = cte.join(Annotation, parent_id=cte.col.id).values(
-            "id", "parent_id", "raw_text"
-        )
-        return base_qs.union(recursive_qs, all=True)
-
-    cte = CTE.recursive(get_full_tree)
-    full_tree_qs = with_cte(cte, select=cte.queryset()).order_by("id")
-    nodes = list(full_tree_qs)
-    full_tree = build_flat_tree(nodes, type_name="AnnotationType", text_key="raw_text")
-    return full_tree
+    nodes = TreeTraversalService.get_nodes(
+        root,
+        info.context.user,
+        mode="full",
+        text_field="raw_text",
+        request=info.context,
+    )
+    return build_flat_tree(nodes, type_name="AnnotationType", text_key="raw_text")
 
 
 def _resolve_AnnotationType_subtree(root, info):
-    """
-    Returns a combined tree that includes:
-    - The path from the root ancestor to this annotation (ancestors).
-    - This annotation and all its descendants.
-    """
-    from django_cte import CTE, with_cte
-
-    # Find all ancestors up to the root
-    ancestors = []
-    node = root
-    while node.parent_id is not None:
-        ancestors.append(node)
-        node = node.parent
-    ancestors.append(node)  # Include the root ancestor
-    ancestor_ids = [ancestor.id for ancestor in ancestors]
-
-    # Get all descendants of the current node
-    def get_descendants(cte):
-        base_qs = Annotation.objects.filter(parent_id=root.id).values(
-            "id", "parent_id", "raw_text"
-        )
-        recursive_qs = cte.join(Annotation, parent_id=cte.col.id).values(
-            "id", "parent_id", "raw_text"
-        )
-        return base_qs.union(recursive_qs, all=True)
-
-    descendants_cte = CTE.recursive(get_descendants)
-    descendants_qs = with_cte(
-        descendants_cte, select=descendants_cte.queryset()
-    ).values("id", "parent_id", "raw_text")
-
-    # Combine ancestors and descendants
-    combined_qs = (
-        Annotation.objects.filter(id__in=ancestor_ids)
-        .values("id", "parent_id", "raw_text")
-        .union(descendants_qs, all=True)
-    )
-
-    subtree_nodes = list(combined_qs)
-    subtree = build_flat_tree(
-        subtree_nodes, type_name="AnnotationType", text_key="raw_text"
+    nodes = TreeTraversalService.get_nodes(
+        root,
+        info.context.user,
+        mode="subtree",
+        text_field="raw_text",
+        request=info.context,
     )
-    return subtree
+    return build_flat_tree(nodes, type_name="AnnotationType", text_key="raw_text")
 
 
 @strawberry.type(name="AnnotationType")
@@ -336,7 +275,10 @@ def long_description(self, info: strawberry.Info) -> str | None:
         return coerce_str(getattr(self, "long_description", None))
 
     json: GenericScalar | None = strawberry.field(name="json", default=None)
-    parent: AnnotationType | None = strawberry.field(name="parent", default=None)
+
+    @strawberry.field(name="parent")
+    def parent(self, info: strawberry.Info) -> AnnotationType | None:
+        return resolve_visible_fk(self, info, "parent_id", "AnnotationType")
 
     @strawberry.field(
         name="annotationType",
@@ -372,23 +314,34 @@ def corpus(
         # private corpus via its ``corpus_id``.
         return resolve_visible_fk(self, info, "corpus_id", "CorpusType")
 
-    analysis: None | (
-        Annotated[AnalysisType, strawberry.lazy("config.graphql.extract_types")]
-    ) = strawberry.field(name="analysis", default=None)
-    created_by_analysis: None | (
-        Annotated[AnalysisType, strawberry.lazy("config.graphql.extract_types")]
-    ) = strawberry.field(
+    @strawberry.field(name="analysis")
+    def analysis(
+        self, info: strawberry.Info
+    ) -> (
+        None | Annotated[AnalysisType, strawberry.lazy("config.graphql.extract_types")]
+    ):
+        return resolve_visible_fk(self, info, "analysis_id", "AnalysisType")
+
+    @strawberry.field(
         name="createdByAnalysis",
         description="If set, this annotation is private to the analysis that created it",
-        default=None,
     )
-    created_by_extract: None | (
-        Annotated[ExtractType, strawberry.lazy("config.graphql.extract_types")]
-    ) = strawberry.field(
+    def created_by_analysis(
+        self, info: strawberry.Info
+    ) -> (
+        None | Annotated[AnalysisType, strawberry.lazy("config.graphql.extract_types")]
+    ):
+        return resolve_visible_fk(self, info, "created_by_analysis_id", "AnalysisType")
+
+    @strawberry.field(
         name="createdByExtract",
         description="If set, this annotation is private to the extract that created it",
-        default=None,
     )
+    def created_by_extract(
+        self, info: strawberry.Info
+    ) -> None | Annotated[ExtractType, strawberry.lazy("config.graphql.extract_types")]:
+        return resolve_visible_fk(self, info, "created_by_extract_id", "ExtractType")
+
     corpus_action: None | (
         Annotated[CorpusActionType, strawberry.lazy("config.graphql.agent_types")]
     ) = strawberry.field(
@@ -1184,9 +1137,14 @@ def label_type(
             getattr(self, "label_type", None),
         )
 
-    analyzer: None | (
-        Annotated[AnalyzerType, strawberry.lazy("config.graphql.extract_types")]
-    ) = strawberry.field(name="analyzer", default=None)
+    @strawberry.field(name="analyzer")
+    def analyzer(
+        self, info: strawberry.Info
+    ) -> (
+        None | Annotated[AnalyzerType, strawberry.lazy("config.graphql.extract_types")]
+    ):
+        return resolve_visible_fk(self, info, "analyzer_id", "AnalyzerType")
+
     read_only: bool = strawberry.field(name="readOnly", default=None)
 
     @strawberry.field(name="color")
@@ -1618,9 +1576,14 @@ def annotation_labels(
             },
         )
 
-    analyzer: None | (
-        Annotated[AnalyzerType, strawberry.lazy("config.graphql.extract_types")]
-    ) = strawberry.field(name="analyzer", default=None)
+    @strawberry.field(name="analyzer")
+    def analyzer(
+        self, info: strawberry.Info
+    ) -> (
+        None | Annotated[AnalyzerType, strawberry.lazy("config.graphql.extract_types")]
+    ):
+        return resolve_visible_fk(self, info, "analyzer_id", "AnalyzerType")
+
     is_default: bool = strawberry.field(name="isDefault", default=None)
 
     @strawberry.field(name="usedByCorpuses")
@@ -1983,26 +1946,42 @@ def target_annotations(
             },
         )
 
-    analyzer: None | (
-        Annotated[AnalyzerType, strawberry.lazy("config.graphql.extract_types")]
-    ) = strawberry.field(name="analyzer", default=None)
-    analysis: None | (
-        Annotated[AnalysisType, strawberry.lazy("config.graphql.extract_types")]
-    ) = strawberry.field(name="analysis", default=None)
-    created_by_analysis: None | (
-        Annotated[AnalysisType, strawberry.lazy("config.graphql.extract_types")]
-    ) = strawberry.field(
+    @strawberry.field(name="analyzer")
+    def analyzer(
+        self, info: strawberry.Info
+    ) -> (
+        None | Annotated[AnalyzerType, strawberry.lazy("config.graphql.extract_types")]
+    ):
+        return resolve_visible_fk(self, info, "analyzer_id", "AnalyzerType")
+
+    @strawberry.field(name="analysis")
+    def analysis(
+        self, info: strawberry.Info
+    ) -> (
+        None | Annotated[AnalysisType, strawberry.lazy("config.graphql.extract_types")]
+    ):
+        return resolve_visible_fk(self, info, "analysis_id", "AnalysisType")
+
+    @strawberry.field(
         name="createdByAnalysis",
         description="If set, this relationship is private to the analysis that created it",
-        default=None,
     )
-    created_by_extract: None | (
-        Annotated[ExtractType, strawberry.lazy("config.graphql.extract_types")]
-    ) = strawberry.field(
+    def created_by_analysis(
+        self, info: strawberry.Info
+    ) -> (
+        None | Annotated[AnalysisType, strawberry.lazy("config.graphql.extract_types")]
+    ):
+        return resolve_visible_fk(self, info, "created_by_analysis_id", "AnalysisType")
+
+    @strawberry.field(
         name="createdByExtract",
         description="If set, this relationship is private to the extract that created it",
-        default=None,
     )
+    def created_by_extract(
+        self, info: strawberry.Info
+    ) -> None | Annotated[ExtractType, strawberry.lazy("config.graphql.extract_types")]:
+        return resolve_visible_fk(self, info, "created_by_extract_id", "ExtractType")
+
     structural: bool = strawberry.field(name="structural", default=None)
     is_public: bool = strawberry.field(name="isPublic", default=None)
     creator: Annotated[UserType, strawberry.lazy("config.graphql.user_types")] = (
@@ -2192,9 +2171,14 @@ def resolution_status(
             getattr(self, "resolution_status", None),
         )
 
-    created_by_analysis: None | (
-        Annotated[AnalysisType, strawberry.lazy("config.graphql.extract_types")]
-    ) = strawberry.field(name="createdByAnalysis", default=None)
+    @strawberry.field(name="createdByAnalysis")
+    def created_by_analysis(
+        self, info: strawberry.Info
+    ) -> (
+        None | Annotated[AnalysisType, strawberry.lazy("config.graphql.extract_types")]
+    ):
+        return resolve_visible_fk(self, info, "created_by_analysis_id", "AnalysisType")
+
     is_provisional: bool = strawberry.field(name="isProvisional", default=None)
 
 
@@ -2215,100 +2199,32 @@ def _resolve_NoteType_revisions(root, info):
 
 
 def _resolve_NoteType_descendants_tree(root, info):
-    """
-    Returns a flat list of descendant notes,
-    each including only the IDs of its immediate children.
-    """
-    from django_cte import CTE, with_cte
-
-    def get_descendants(cte):
-        base_qs = Note.objects.filter(parent_id=root.id).values(
-            "id", "parent_id", "content"
-        )
-        recursive_qs = cte.join(Note, parent_id=cte.col.id).values(
-            "id", "parent_id", "content"
-        )
-        return base_qs.union(recursive_qs, all=True)
-
-    cte = CTE.recursive(get_descendants)
-    descendants_qs = with_cte(cte, select=cte.queryset()).order_by("id")
-    descendants_list = list(descendants_qs)
-    descendants_tree = build_flat_tree(
-        descendants_list, type_name="NoteType", text_key="content"
+    nodes = TreeTraversalService.get_nodes(
+        root,
+        info.context.user,
+        mode="descendants",
+        text_field="content",
+        request=info.context,
     )
-    return descendants_tree
+    return build_flat_tree(nodes, type_name="NoteType", text_key="content")
 
 
 def _resolve_NoteType_full_tree(root, info):
-    """
-    Returns a flat list of notes from the root ancestor,
-    each including only the IDs of its immediate children.
-    """
-    from django_cte import CTE, with_cte
-
-    # Find the root ancestor
-    tree_root = root
-    while tree_root.parent_id is not None:
-        tree_root = tree_root.parent
-
-    def get_full_tree(cte):
-        base_qs = Note.objects.filter(id=tree_root.id).values(
-            "id", "parent_id", "content"
-        )
-        recursive_qs = cte.join(Note, parent_id=cte.col.id).values(
-            "id", "parent_id", "content"
-        )
-        return base_qs.union(recursive_qs, all=True)
-
-    cte = CTE.recursive(get_full_tree)
-    full_tree_qs = with_cte(cte, select=cte.queryset()).order_by("id")
-    nodes = list(full_tree_qs)
-    full_tree = build_flat_tree(nodes, type_name="NoteType", text_key="content")
-    return full_tree
+    nodes = TreeTraversalService.get_nodes(
+        root, info.context.user, mode="full", text_field="content", request=info.context
+    )
+    return build_flat_tree(nodes, type_name="NoteType", text_key="content")
 
 
 def _resolve_NoteType_subtree(root, info):
-    """
-    Returns a combined tree that includes:
-    - The path from the root ancestor to this note (ancestors).
-    - This note and all its descendants.
-    """
-    from django_cte import CTE, with_cte
-
-    # Find all ancestors up to the root
-    ancestors = []
-    node = root
-    while node.parent_id is not None:
-        ancestors.append(node)
-        node = node.parent
-    ancestors.append(node)  # Include the root ancestor
-    ancestor_ids = [ancestor.id for ancestor in ancestors]
-
-    # Get all descendants of the current node
-    def get_descendants(cte):
-        base_qs = Note.objects.filter(parent_id=root.id).values(
-            "id", "parent_id", "content"
-        )
-        recursive_qs = cte.join(Note, parent_id=cte.col.id).values(
-            "id", "parent_id", "content"
-        )
-        return base_qs.union(recursive_qs, all=True)
-
-    descendants_cte = CTE.recursive(get_descendants)
-    descendants_qs = with_cte(
-        descendants_cte, select=descendants_cte.queryset()
-    ).values("id", "parent_id", "content")
-
-    # Combine ancestors and descendants
-    combined_qs = (
-        Note.objects.filter(id__in=ancestor_ids)
-        .values("id", "parent_id", "content")
-        .union(descendants_qs, all=True)
+    nodes = TreeTraversalService.get_nodes(
+        root,
+        info.context.user,
+        mode="subtree",
+        text_field="content",
+        request=info.context,
     )
-
-    subtree_nodes = list(combined_qs)
-    subtree = build_flat_tree(subtree_nodes, type_name="NoteType", text_key="content")
-    return subtree
+    return build_flat_tree(nodes, type_name="NoteType", text_key="content")
 
 
 def _resolve_NoteType_current_version(root, info):
@@ -2342,16 +2258,24 @@ def title(self, info: strawberry.Info) -> str:
     def content(self, info: strawberry.Info) -> str:
         return coerce_str(getattr(self, "content", None))
 
-    parent: NoteType | None = strawberry.field(name="parent", default=None)
-    corpus: None | (
-        Annotated[CorpusType, strawberry.lazy("config.graphql.corpus_types")]
-    ) = strawberry.field(name="corpus", default=None)
+    @strawberry.field(name="parent")
+    def parent(self, info: strawberry.Info) -> NoteType | None:
+        return resolve_visible_fk(self, info, "parent_id", "NoteType")
+
+    @strawberry.field(name="corpus")
+    def corpus(
+        self, info: strawberry.Info
+    ) -> None | Annotated[CorpusType, strawberry.lazy("config.graphql.corpus_types")]:
+        return resolve_visible_fk(self, info, "corpus_id", "CorpusType")
+
     document: Annotated[
         DocumentType, strawberry.lazy("config.graphql.document_types")
     ] = strawberry.field(name="document", default=None)
-    annotation: AnnotationType | None = strawberry.field(
-        name="annotation", default=None
-    )
+
+    @strawberry.field(name="annotation")
+    def annotation(self, info: strawberry.Info) -> AnnotationType | None:
+        return resolve_visible_fk(self, info, "annotation_id", "AnnotationType")
+
     is_public: bool = strawberry.field(name="isPublic", default=None)
     creator: Annotated[UserType, strawberry.lazy("config.graphql.user_types")] = (
         strawberry.field(name="creator", default=None)
@@ -2767,13 +2691,17 @@ def last_error(self, info: strawberry.Info) -> str | None:
         description="Per-corpus demand breakdown: [{corpus_id, mention_count, top_detection_tier}].",
         default=None,
     )
-    ingested_document: None | (
-        Annotated[DocumentType, strawberry.lazy("config.graphql.document_types")]
-    ) = strawberry.field(
+
+    @strawberry.field(
         name="ingestedDocument",
         description="The Document imported for this key once ingested (else null).",
-        default=None,
     )
+    def ingested_document(
+        self, info: strawberry.Info
+    ) -> (
+        None | Annotated[DocumentType, strawberry.lazy("config.graphql.document_types")]
+    ):
+        return resolve_visible_fk(self, info, "ingested_document_id", "DocumentType")
 
     @strawberry.field(
         name="ingestable",
diff --git a/config/graphql/conversation_types.py b/config/graphql/conversation_types.py
index 47acd63b68..b632cc2e44 100644
--- a/config/graphql/conversation_types.py
+++ b/config/graphql/conversation_types.py
@@ -1180,11 +1180,11 @@ def agent_configuration(
         kwargs = strip_unset({})
         return _resolve_MessageType_agent_configuration(self, info, **kwargs)
 
-    parent_message: MessageType | None = strawberry.field(
-        name="parentMessage",
-        description="Parent message for threaded replies",
-        default=None,
+    @strawberry.field(
+        name="parentMessage", description="Parent message for threaded replies"
     )
+    def parent_message(self, info: strawberry.Info) -> MessageType | None:
+        return resolve_visible_fk(self, info, "parent_message_id", "MessageType")
 
     @strawberry.field(
         name="content", description="The textual content of the chat message"
@@ -2018,9 +2018,9 @@ class ModerationActionType(Node):
     def conversation(self, info: strawberry.Info) -> ConversationType | None:
         return resolve_visible_fk(self, info, "conversation_id", "ConversationType")
 
-    message: MessageType | None = strawberry.field(
-        name="message", description="The message that was moderated", default=None
-    )
+    @strawberry.field(name="message", description="The message that was moderated")
+    def message(self, info: strawberry.Info) -> MessageType | None:
+        return resolve_visible_fk(self, info, "message_id", "MessageType")
 
     @strawberry.field(name="actionType", description="Type of moderation action taken")
     def action_type(
diff --git a/config/graphql/core/mutations.py b/config/graphql/core/mutations.py
index 7042391008..5878c7a9d5 100644
--- a/config/graphql/core/mutations.py
+++ b/config/graphql/core/mutations.py
@@ -11,8 +11,9 @@
 import logging
 import traceback
 from collections.abc import Sequence
-from typing import Any
+from typing import Any, cast
 
+from django.db.models import Model
 from rest_framework import serializers
 
 from config.graphql.core.auth import PermissionDenied
@@ -95,14 +96,23 @@ def _drf_mutation_body(
     obj_id = None
 
     try:
-        if info.context.user:
-            kwargs["creator"] = info.context.user.id
-        else:
+        if not info.context.user:
             raise ValueError("No user in this request...")
 
+        # Ownership grants management permissions. Editing a shared object
+        # must never turn an UPDATE grantee into its owner.
+        is_update = lookup_field in kwargs
+        if is_update:
+            kwargs.pop("creator", None)
+            kwargs.pop("creator_id", None)
+        else:
+            kwargs["creator"] = info.context.user.id
+
         for pk_field in pk_fields:
             if pk_field in kwargs:
                 raw_value = kwargs[pk_field]
+                if raw_value is None:
+                    continue
                 if isinstance(raw_value, list):
                     kwargs[pk_field] = [
                         from_global_id(global_id)[1] for global_id in raw_value
@@ -110,7 +120,25 @@ def _drf_mutation_body(
                 else:
                     kwargs[pk_field] = from_global_id(raw_value)[1]
 
-        is_update = lookup_field in kwargs
+                # A writable parent is not authority to attach someone else's
+                # private label set or label (and expose its intrinsic fields).
+                # Categories are install-wide vocabulary, not private data.
+                if pk_field in {"label_set", "annotation_label"}:
+                    related_model = (
+                        cast(type[Model], model)._meta.get_field(pk_field).related_model
+                    )
+                    if (
+                        BaseService.get_or_none(
+                            related_model,
+                            kwargs[pk_field],
+                            info.context.user,
+                            request=info.context,
+                        )
+                        is None
+                    ):
+                        raise serializers.ValidationError(
+                            {pk_field: "Resource not found or access denied."}
+                        )
 
         if is_update:
             lookup_pk = from_global_id(kwargs[lookup_field])[1]
diff --git a/config/graphql/corpus_types.py b/config/graphql/corpus_types.py
index 9c1932294e..49bb0c74f7 100644
--- a/config/graphql/corpus_types.py
+++ b/config/graphql/corpus_types.py
@@ -119,18 +119,19 @@ def _resolve_CorpusType_label_set(root, info):
     to copy those annotations to the label_set instance so that its
     count resolvers can use them instead of hitting the database.
     """
-    if root.label_set is None:
+    label_set = resolve_visible_fk(root, info, "label_set_id", "LabelSetType")
+    if label_set is None:
         return None
 
-    # Copy annotated counts to the label_set instance
-    if hasattr(root, "_label_doc_count"):
-        root.label_set._doc_label_count = root._label_doc_count
-    if hasattr(root, "_label_span_count"):
-        root.label_set._span_label_count = root._label_span_count
-    if hasattr(root, "_label_token_count"):
-        root.label_set._token_label_count = root._label_token_count
-
-    return root.label_set
+    # Copy annotated counts to the visible label set instance.
+    for source, target in (
+        ("_label_doc_count", "_doc_label_count"),
+        ("_label_span_count", "_span_label_count"),
+        ("_label_token_count", "_token_label_count"),
+    ):
+        if hasattr(root, source):
+            setattr(label_set, target, getattr(root, source))
+    return label_set
 
 
 def _resolve_CorpusType_engagement_metrics(root, info):
@@ -1308,9 +1309,13 @@ def analyses(
             node_type_name="AnalysisType",
         )
 
-    metadata_schema: None | (
-        Annotated[FieldsetType, strawberry.lazy("config.graphql.extract_types")]
-    ) = strawberry.field(name="metadataSchema", default=None)
+    @strawberry.field(name="metadataSchema")
+    def metadata_schema(
+        self, info: strawberry.Info
+    ) -> (
+        None | Annotated[FieldsetType, strawberry.lazy("config.graphql.extract_types")]
+    ):
+        return resolve_visible_fk(self, info, "metadata_schema_id", "FieldsetType")
 
     @strawberry.field(name="extracts")
     def extracts(
diff --git a/config/graphql/document_types.py b/config/graphql/document_types.py
index c0162bc837..5b8463e676 100644
--- a/config/graphql/document_types.py
+++ b/config/graphql/document_types.py
@@ -1431,9 +1431,11 @@ def summary_revisions(
         kwargs = strip_unset({"corpus_id": corpus_id})
         return _resolve_DocumentType_summary_revisions(self, info, **kwargs)
 
-    memory_for_corpus: None | (
-        Annotated[CorpusType, strawberry.lazy("config.graphql.corpus_types")]
-    ) = strawberry.field(name="memoryForCorpus", default=None)
+    @strawberry.field(name="memoryForCorpus")
+    def memory_for_corpus(
+        self, info: strawberry.Info
+    ) -> None | Annotated[CorpusType, strawberry.lazy("config.graphql.corpus_types")]:
+        return resolve_visible_fk(self, info, "memory_for_corpus_id", "CorpusType")
 
     @strawberry.field(
         name="corpusActionExecutions",
@@ -2784,12 +2786,19 @@ def data(
             node_type_name="DatacellType",
         )
 
-    analysis: None | (
-        Annotated[AnalysisType, strawberry.lazy("config.graphql.extract_types")]
-    ) = strawberry.field(name="analysis", default=None)
-    extract: None | (
-        Annotated[ExtractType, strawberry.lazy("config.graphql.extract_types")]
-    ) = strawberry.field(name="extract", default=None)
+    @strawberry.field(name="analysis")
+    def analysis(
+        self, info: strawberry.Info
+    ) -> (
+        None | Annotated[AnalysisType, strawberry.lazy("config.graphql.extract_types")]
+    ):
+        return resolve_visible_fk(self, info, "analysis_id", "AnalysisType")
+
+    @strawberry.field(name="extract")
+    def extract(
+        self, info: strawberry.Info
+    ) -> None | Annotated[ExtractType, strawberry.lazy("config.graphql.extract_types")]:
+        return resolve_visible_fk(self, info, "extract_id", "ExtractType")
 
     @strawberry.field(name="myPermissions")
     def my_permissions(self, info: strawberry.Info) -> GenericScalar | None:
@@ -2853,9 +2862,13 @@ def relationship_type(
             AnnotationLabelType, strawberry.lazy("config.graphql.annotation_types")
         ]
     ) = strawberry.field(name="annotationLabel", default=None)
-    corpus: None | (
-        Annotated[CorpusType, strawberry.lazy("config.graphql.corpus_types")]
-    ) = strawberry.field(name="corpus", default=None)
+
+    @strawberry.field(name="corpus")
+    def corpus(
+        self, info: strawberry.Info
+    ) -> None | Annotated[CorpusType, strawberry.lazy("config.graphql.corpus_types")]:
+        return resolve_visible_fk(self, info, "corpus_id", "CorpusType")
+
     data: GenericScalar | None = strawberry.field(name="data", default=None)
 
     @strawberry.field(name="myPermissions")
diff --git a/config/graphql/extract_types.py b/config/graphql/extract_types.py
index da09542909..b0dcbfe539 100644
--- a/config/graphql/extract_types.py
+++ b/config/graphql/extract_types.py
@@ -51,6 +51,7 @@
 from opencontractserver.extracts.models import Column, Datacell, Extract, Fieldset
 from opencontractserver.notifications.models import Notification
 from opencontractserver.shared.services.base import BaseService
+from opencontractserver.types.enums import PermissionTypes
 from opencontractserver.utils.ids import from_global_id
 
 
@@ -124,9 +125,11 @@ def icon(self, info: strawberry.Info) -> str:
         kwargs = strip_unset({})
         return _resolve_AnalyzerType_icon(self, info, **kwargs)
 
-    host_gremlin: GremlinEngineType_WRITE | None = strawberry.field(
-        name="hostGremlin", default=None
-    )
+    @strawberry.field(name="hostGremlin")
+    def host_gremlin(self, info: strawberry.Info) -> GremlinEngineType_WRITE | None:
+        return resolve_visible_fk(
+            self, info, "host_gremlin_id", "GremlinEngineType_WRITE"
+        )
 
     @strawberry.field(name="taskName")
     def task_name(self, info: strawberry.Info) -> str | None:
@@ -529,6 +532,11 @@ def analyzer_set(
 
     @strawberry.field(name="apiKey")
     def api_key(self, info: strawberry.Info) -> str | None:
+        # READ (including publication) never grants access to credentials.
+        if not BaseService.user_has(
+            self, info.context.user, PermissionTypes.UPDATE, request=info.context
+        ):
+            return None
         return coerce_str(getattr(self, "api_key", None))
 
     @strawberry.field(name="myPermissions")
@@ -544,7 +552,18 @@ def object_shared_with(self, info: strawberry.Info) -> GenericScalar | None:
         return core_permissions.resolve_object_shared_with(self, info)
 
 
-register_type("GremlinEngineType_WRITE", GremlinEngineType_WRITE, model=GremlinEngine)
+def _get_node_GremlinEngineType_WRITE(info, pk):
+    return BaseService.get_or_none(
+        GremlinEngine, pk, info.context.user, request=info.context
+    )
+
+
+register_type(
+    "GremlinEngineType_WRITE",
+    GremlinEngineType_WRITE,
+    model=GremlinEngine,
+    get_node=_get_node_GremlinEngineType_WRITE,
+)
 
 
 GremlinEngineType_WRITEConnection = make_connection_types(
@@ -737,11 +756,14 @@ def error(self, info: strawberry.Info) -> str | None:
     corpus_action: None | (
         Annotated[CorpusActionType, strawberry.lazy("config.graphql.agent_types")]
     ) = strawberry.field(name="corpusAction", default=None)
-    parent_extract: ExtractType | None = strawberry.field(
+
+    @strawberry.field(
         name="parentExtract",
         description="Extract this iteration was forked from. Null for the root of an iteration series.",
-        default=None,
     )
+    def parent_extract(self, info: strawberry.Info) -> ExtractType | None:
+        return resolve_visible_fk(self, info, "parent_extract_id", "ExtractType")
+
     model_config: GenericScalar | None = strawberry.field(
         name="modelConfig",
         description="Captured model/run configuration for this iteration.",
@@ -1710,7 +1732,11 @@ class DatacellType(Node):
     )
     created: datetime.datetime = strawberry.field(name="created", default=None)
     modified: datetime.datetime = strawberry.field(name="modified", default=None)
-    extract: ExtractType | None = strawberry.field(name="extract", default=None)
+
+    @strawberry.field(name="extract")
+    def extract(self, info: strawberry.Info) -> ExtractType | None:
+        return resolve_visible_fk(self, info, "extract_id", "ExtractType")
+
     column: ColumnType = strawberry.field(name="column", default=None)
     document: Annotated[
         DocumentType, strawberry.lazy("config.graphql.document_types")
diff --git a/config/graphql/research_types.py b/config/graphql/research_types.py
index b26a88b495..5734d20837 100644
--- a/config/graphql/research_types.py
+++ b/config/graphql/research_types.py
@@ -351,13 +351,17 @@ def conversation(
     ):
         return resolve_visible_fk(self, info, "conversation_id", "ConversationType")
 
-    originating_message: None | (
-        Annotated[MessageType, strawberry.lazy("config.graphql.conversation_types")]
-    ) = strawberry.field(
+    @strawberry.field(
         name="originatingMessage",
         description="User chat message that triggered this run, if any",
-        default=None,
     )
+    def originating_message(
+        self, info: strawberry.Info
+    ) -> (
+        None
+        | Annotated[MessageType, strawberry.lazy("config.graphql.conversation_types")]
+    ):
+        return resolve_visible_fk(self, info, "originating_message_id", "MessageType")
 
     @strawberry.field(
         name="workspaceDocument",
diff --git a/config/graphql/security.py b/config/graphql/security.py
index 5bc031d275..51d9ca8381 100644
--- a/config/graphql/security.py
+++ b/config/graphql/security.py
@@ -268,12 +268,18 @@ def _measure_depth(
     current_depth: int = 0,
     context: Any = None,
     visited_fragments: set[str] | None = None,
+    fragment_depths: dict[str, int] | None = None,
 ) -> int:
     """Recursively measure the maximum depth of selection sets.
 
     Follows fragment spreads through the fragment registry to prevent
     attackers from hiding depth behind named fragments.
     """
+    # Cache relative fragment depth, not whether it appeared anywhere in
+    # the query. Reuse at a deeper location must count its full depth, while
+    # a fragment DAG must not expand exponentially during validation.
+    if fragment_depths is None:
+        fragment_depths = {}
     if visited_fragments is None:
         visited_fragments = set()
     if not hasattr(node, "selection_set") or node.selection_set is None:
@@ -283,25 +289,35 @@ def _measure_depth(
     for selection in node.selection_set.selections:
         if isinstance(selection, ast.FieldNode):
             child_depth = _measure_depth(
-                selection, current_depth + 1, context, visited_fragments
+                selection,
+                current_depth + 1,
+                context,
+                visited_fragments,
+                fragment_depths,
             )
         elif isinstance(selection, ast.InlineFragmentNode):
             child_depth = _measure_depth(
-                selection, current_depth, context, visited_fragments
+                selection, current_depth, context, visited_fragments, fragment_depths
             )
         elif isinstance(selection, ast.FragmentSpreadNode) and context is not None:
             frag_name = selection.name.value
-            if frag_name not in visited_fragments:
-                visited_fragments.add(frag_name)
-                fragment = context.get_fragment(frag_name)
-                if fragment:
-                    child_depth = _measure_depth(
-                        fragment, current_depth, context, visited_fragments
-                    )
-                else:
-                    child_depth = current_depth
+            if frag_name in visited_fragments:
+                child_depth = current_depth  # Cycles are rejected by spec validation.
             else:
-                child_depth = current_depth  # cycle guard
+                if frag_name not in fragment_depths:
+                    fragment = context.get_fragment(frag_name)
+                    fragment_depths[frag_name] = (
+                        _measure_depth(
+                            fragment,
+                            0,
+                            context,
+                            visited_fragments | {frag_name},
+                            fragment_depths,
+                        )
+                        if fragment
+                        else 0
+                    )
+                child_depth = current_depth + fragment_depths[frag_name]
         else:
             child_depth = current_depth
         if child_depth > max_child:
diff --git a/config/graphql_api_token_auth/backends.py b/config/graphql_api_token_auth/backends.py
index a6c7af0c55..6220a5f94b 100644
--- a/config/graphql_api_token_auth/backends.py
+++ b/config/graphql_api_token_auth/backends.py
@@ -61,7 +61,6 @@ def authenticate(
             return None
 
         auth = get_authorization_header(request).split()
-        logger.debug(f"Authorization header: {auth}")
 
         if not auth or auth[0].lower() != settings.API_TOKEN_PREFIX.lower().encode():
             logger.debug("Invalid or missing auth prefix")
@@ -95,7 +94,7 @@ def authenticate_credentials(self, key: str) -> AbstractBaseUser:
             token = model.objects.select_related("user").get(key=key)
             logger.debug(f"Found token for user: {token.user.username}")
         except model.DoesNotExist:
-            logger.warning(f"Authentication failed: Invalid token {key[:8]}...")
+            logger.warning("Authentication failed: Invalid token")
             raise exceptions.AuthenticationFailed(_("Invalid token."))
 
         if not token.user.is_active:
diff --git a/config/graphql_auth0_auth/backends.py b/config/graphql_auth0_auth/backends.py
index dd7faf8f77..e7309a17de 100644
--- a/config/graphql_auth0_auth/backends.py
+++ b/config/graphql_auth0_auth/backends.py
@@ -33,9 +33,6 @@ def authenticate(self, request=None, **kwargs):
         logger.debug(
             f"Auth0RemoteUserJSONWebTokenBackend.authenticate() - Starting with request: {request}"
         )
-        logger.debug(
-            f"Auth0RemoteUserJSONWebTokenBackend.authenticate() - kwargs: {kwargs}"
-        )
 
         if request is None or getattr(request, "_jwt_token_auth", False):
             logger.debug(
@@ -47,10 +44,6 @@ def authenticate(self, request=None, **kwargs):
         logger.debug(
             f"Auth0RemoteUserJSONWebTokenBackend.authenticate() - token retrieved: {'Present' if token else 'None'}"
         )
-        if token:
-            logger.debug(
-                f"Auth0RemoteUserJSONWebTokenBackend.authenticate() - token first 10 chars: {token[:10]}"
-            )
 
         if token is not None:
             try:
diff --git a/config/graphql_auth0_auth/utils.py b/config/graphql_auth0_auth/utils.py
index 781b3e44f2..4ed34f81dc 100644
--- a/config/graphql_auth0_auth/utils.py
+++ b/config/graphql_auth0_auth/utils.py
@@ -12,7 +12,6 @@
 
 from config.graphql_auth0_auth.settings import auth0_settings
 from config.jwt_auth import exceptions
-from opencontractserver.constants import TOKEN_LOG_PREFIX_LENGTH
 
 logger = logging.getLogger(__name__)
 
@@ -102,11 +101,7 @@ def _can_serve_stale(current_time: float) -> bool:
 
 
 def jwt_auth0_decode(token):
-    logger.debug(
-        "jwt_auth0_decode() - Attempting to decode token, first %d chars: %s...",
-        TOKEN_LOG_PREFIX_LENGTH,
-        token[:TOKEN_LOG_PREFIX_LENGTH],
-    )
+    logger.debug("Processing Auth0 token")
     try:
         header = jwt.get_unverified_header(token)
         logger.debug("jwt_auth0_decode() - Header: %s", header)
@@ -166,11 +161,7 @@ def jwt_auth0_decode(token):
 
 
 def get_payload(token):
-    logger.debug(
-        "get_payload() - Processing token, first %d chars: %s...",
-        TOKEN_LOG_PREFIX_LENGTH,
-        token[:TOKEN_LOG_PREFIX_LENGTH] if token else "None",
-    )
+    logger.debug("Processing Auth0 token")
     try:
         payload = auth0_settings.AUTH0_DECODE_HANDLER(token)
         logger.debug(
@@ -602,11 +593,7 @@ def get_user_by_token(token, **kwargs):
     user exists and settings is set to create user obj for unknown user,
     create a user, configure it, and return user obj
     """
-    logger.debug(
-        "get_user_by_token() - Starting with token first %d chars: %s...",
-        TOKEN_LOG_PREFIX_LENGTH,
-        token[:TOKEN_LOG_PREFIX_LENGTH] if token else "None",
-    )
+    logger.debug("Processing Auth0 token")
     try:
         payload = get_payload(token)
         logger.debug(
diff --git a/config/jwt_utils.py b/config/jwt_utils.py
index a7bac054c1..69f039f4f5 100644
--- a/config/jwt_utils.py
+++ b/config/jwt_utils.py
@@ -67,7 +67,7 @@ def _validate_graphql_jwt_token(token: str) -> "User":
     from config.jwt_auth.exceptions import JSONWebTokenError
     from config.jwt_auth.utils import get_payload, get_user_by_payload
 
-    logger.debug(f"Validating graphql_jwt token: {token[:10]}...")
+    logger.debug("Validating local JWT")
 
     # get_payload raises JSONWebTokenExpired or JSONWebTokenError
     payload = get_payload(token)
@@ -102,7 +102,7 @@ def _validate_auth0_token(token: str) -> "User":
     from config.graphql_auth0_auth.utils import get_user_by_token
     from config.jwt_auth.exceptions import JSONWebTokenError
 
-    logger.debug(f"Validating Auth0 token: {token[:10]}...")
+    logger.debug("Validating Auth0 JWT")
 
     # get_user_by_token handles payload extraction, user lookup/creation,
     # and raises JSONWebTokenExpired or JSONWebTokenError as appropriate
diff --git a/config/websocket/auth_handshake.py b/config/websocket/auth_handshake.py
index 924342e401..ea4ee25f37 100644
--- a/config/websocket/auth_handshake.py
+++ b/config/websocket/auth_handshake.py
@@ -14,10 +14,11 @@
         "EXPIRED" | "INVALID" | "USER_MISMATCH" | "PERMISSION_REVOKED"}
     {"type": "AUTH_REFRESH_REQUIRED", "grace_seconds": float}
 
-Security guarantees enforced by handle_auth_message():
+Security guarantees enforced at dispatch, during refresh, and by a watchdog:
   1. A live socket bound to user A cannot be re-bound to user B (USER_MISMATCH).
-  2. If the user has lost access to a bound resource since connect, the next
-     AUTH frame closes 4003 (PERMISSION_REVOKED).
+  2. Resource revocation closes 4003 (PERMISSION_REVOKED) on the next event
+     or periodic recheck. Streaming tokens share a recheck for at most 1 second;
+     token expiry is checked on every event regardless of that window.
   3. An expired/invalid AUTH frame closes the socket (4001/4002) and never
      leaves the consumer in an inconsistent state.
 """
@@ -39,6 +40,7 @@
     WS_CLOSE_PERMISSION_DENIED,
     WS_CLOSE_TOKEN_EXPIRED,
     WS_CLOSE_TOKEN_INVALID,
+    verified_token_expiry,
 )
 from opencontractserver.constants.auth import WS_AUTH_REFRESH_GRACE_SECONDS
 
@@ -49,6 +51,9 @@
 # floor cannot interfere with legitimate refreshes but stops a malicious client
 # from spamming AUTH frames to burn DB queries (issue raised in PR #1502 review).
 _MIN_AUTH_FRAME_INTERVAL_SEC = 1.0
+_AUTH_RECHECK_INTERVAL_SEC = 30.0
+# Bound DB work during token fan-out without extending the window per token.
+_STREAM_AUTH_RECHECK_INTERVAL_SEC = 1.0
 
 
 @database_sync_to_async
@@ -70,8 +75,12 @@ class AuthHandshakeMixin:
          code that catches a JSONWebTokenExpired mid-flight.
     """
 
+    # Supplied by AsyncWebsocketConsumer when this cooperative mixin is used.
+    scope: dict[str, Any]
+
     # Populated by accept_with_auth() and updated by handle_auth_message().
     _refresh_grace_task: asyncio.Task | None = None
+    _authorization_task: asyncio.Task | None = None
     _initial_auth_sent: bool = False
     # Tracks whether the handshake has accepted but not yet been cleaned up.
     # The grace-timer guard uses this to avoid calling close() on a socket
@@ -80,10 +89,11 @@ class AuthHandshakeMixin:
     # Monotonic timestamp of the last AUTH frame we accepted; used to throttle
     # spam at the per-connection level before any DB work runs.
     _last_auth_frame_at: float = 0.0
+    _last_authorized_at: float | None = None
 
     @property
-    def current_user(self):
-        return self.scope.get("user")  # type: ignore[attr-defined]
+    def current_user(self) -> Any:
+        return self.scope.get("user") or AnonymousUser()
 
     # ------------------------------------------------------------------ #
     #  Connection accept
@@ -91,10 +101,90 @@ def current_user(self):
 
     async def accept_with_auth(self) -> None:
         """Accept the connection echoing the negotiated subprotocol."""
-        subprotocol = self.scope.get("accepted_subprotocol")  # type: ignore[attr-defined]
+        subprotocol = self.scope.get("accepted_subprotocol")
         await self.accept(subprotocol=subprotocol)  # type: ignore[attr-defined]
         self._handshake_connected = True
+        self._last_authorized_at = None
         await self._send_initial_auth_ok()
+        self._start_authorization_watchdog()
+
+    async def dispatch(self, message: dict[str, Any]) -> None:
+        """Authorize actions and channel broadcasts, even without client AUTH.
+
+        A valid handshake is not a lifetime grant. Revalidation uses fresh
+        users/resources, never the consumer's retained permission caches.
+        """
+        if message["type"] not in ("websocket.connect", "websocket.disconnect"):
+            is_auth = False
+            if message["type"] == "websocket.receive":
+                try:
+                    payload = json.loads(message.get("text") or "null")
+                    is_auth = (
+                        isinstance(payload, dict) and payload.get("type") == "AUTH"
+                    )
+                except (ValueError, TypeError):
+                    # Malformed frames still require authorization before receive().
+                    pass
+            if not self._handshake_connected:
+                return
+            # Only this server-originated channel event may reuse a check.
+            # Client frames and other broadcasts always revalidate fully.
+            if not is_auth and not await self.ensure_authorized(
+                allow_recent=message["type"] == "agent_stream_token"
+            ):
+                return
+        await super().dispatch(message)  # type: ignore[misc]
+
+    async def ensure_authorized(self, *, allow_recent: bool = False) -> bool:
+        """Fail closed on expiry, deactivation, or resource revocation."""
+        user = self.current_user
+        token = self.scope.get("auth_token")
+        checked_at = time.monotonic()
+        try:
+            expires_at = self.scope.get("auth_expires_at")
+            if expires_at is not None and time.time() >= expires_at:
+                raise JSONWebTokenExpired("Token has expired")
+            if allow_recent and self._last_authorized_at is not None:
+                elapsed = checked_at - self._last_authorized_at
+                if 0 <= elapsed < _STREAM_AUTH_RECHECK_INTERVAL_SEC:
+                    return True
+            if token:
+                user = await _get_user_from_token(token)
+            if not await self._validate_resource_permissions(user):
+                await self._fail_auth("PERMISSION_REVOKED", WS_CLOSE_PERMISSION_DENIED)
+                return False
+        except JSONWebTokenExpired:
+            await self._fail_auth("EXPIRED", WS_CLOSE_TOKEN_EXPIRED)
+            return False
+        except JSONWebTokenError:
+            await self._fail_auth("INVALID", WS_CLOSE_TOKEN_INVALID)
+            return False
+        except Exception:
+            logger.exception("WebSocket authorization recheck failed")
+            await self._fail_auth("PERMISSION_REVOKED", WS_CLOSE_PERMISSION_DENIED)
+            return False
+        self.scope["user"] = user
+        self._last_authorized_at = checked_at
+        return True
+
+    def _start_authorization_watchdog(self) -> None:
+        if self._authorization_task is not None:
+            self._authorization_task.cancel()
+        self._authorization_task = asyncio.create_task(self._watch_authorization())
+
+    async def _watch_authorization(self) -> None:
+        """Expire idle/streaming sockets without relying on client cooperation."""
+        try:
+            while self._handshake_connected:
+                delay = _AUTH_RECHECK_INTERVAL_SEC
+                expires_at = self.scope.get("auth_expires_at")
+                if expires_at is not None:
+                    delay = min(delay, max(0.01, expires_at - time.time()))
+                await asyncio.sleep(delay)
+                if not self._handshake_connected or not await self.ensure_authorized():
+                    return
+        except asyncio.CancelledError:
+            return
 
     async def _send_initial_auth_ok(self) -> None:
         if self._initial_auth_sent:
@@ -148,6 +238,7 @@ async def handle_auth_message(self, payload: dict[str, Any]) -> None:
 
         try:
             new_user = await _get_user_from_token(token)
+            expiry = verified_token_expiry(token)
         except JSONWebTokenExpired:
             await self._fail_auth("EXPIRED", WS_CLOSE_TOKEN_EXPIRED)
             return
@@ -188,8 +279,12 @@ async def handle_auth_message(self, payload: dict[str, Any]) -> None:
             return
 
         # Success — swap, ack, cancel any pending grace timer.
-        self.scope["user"] = new_user  # type: ignore[attr-defined]
+        self.scope["user"] = new_user
+        self.scope["auth_token"] = token
+        self.scope["auth_expires_at"] = expiry
+        self._last_authorized_at = None
         self._cancel_refresh_grace_timer()
+        self._start_authorization_watchdog()
         await self.send(  # type: ignore[attr-defined]
             text_data=json.dumps(
                 {
@@ -210,6 +305,11 @@ async def _validate_resource_permissions(self, user) -> bool:
         return True
 
     async def _fail_auth(self, reason: str, close_code: int) -> None:
+        self._handshake_connected = False
+        self._last_authorized_at = None
+        # UnifiedAgentConsumer checks this flag between streaming events.
+        if hasattr(self, "_is_connected"):
+            self._is_connected = False
         try:
             await self.send(  # type: ignore[attr-defined]
                 text_data=json.dumps(
@@ -269,4 +369,10 @@ def _cancel_refresh_grace_timer(self) -> None:
     async def cleanup_auth_handshake(self) -> None:
         """Consumers should call this from their ``disconnect()``."""
         self._handshake_connected = False
+        self._last_authorized_at = None
         self._cancel_refresh_grace_timer()
+        task = self._authorization_task
+        self._authorization_task = None
+        if task is not None and task is not asyncio.current_task():
+            task.cancel()
+            await asyncio.gather(task, return_exceptions=True)
diff --git a/config/websocket/consumers/thread_updates.py b/config/websocket/consumers/thread_updates.py
index b2babfbdfa..a7f3588ae2 100644
--- a/config/websocket/consumers/thread_updates.py
+++ b/config/websocket/consumers/thread_updates.py
@@ -30,8 +30,7 @@
 from config.websocket.middleware import WS_CLOSE_RATE_LIMITED
 from config.websocket.utils.auth_helpers import check_auth_and_close_if_failed
 from opencontractserver.conversations.models import Conversation
-from opencontractserver.corpuses.models import Corpus
-from opencontractserver.documents.models import Document
+from opencontractserver.conversations.services import ConversationService
 from opencontractserver.utils.ids import from_global_id
 
 logger = logging.getLogger(__name__)
@@ -87,7 +86,7 @@ async def connect(self) -> None:
         if await check_auth_and_close_if_failed(self, self.session_id):
             return
 
-        user = self.scope.get("user")
+        user = self.current_user
         self.user_id = user.pk
 
         # Parse query parameters
@@ -276,10 +275,9 @@ def _check_conversation_access(self, user, *, cache_conversation: bool) -> bool:
         """
         Check if ``user`` may participate in ``self.conversation_id``.
 
-        For conversations with BOTH chat_with_corpus AND chat_with_document set
-        (doc-in-corpus threads), the user must have access to BOTH the corpus
-        AND the document (AND logic). For single-context conversations the user
-        only needs access to that one resource.
+        Use the canonical policy: private CHATs require their own authority;
+        THREADs can inherit parent visibility, with both contexts required for
+        document-in-corpus threads. Administrators have no blanket read bypass.
 
         ``cache_conversation`` controls whether the resolved Conversation is
         attached to ``self.conversation`` for downstream use; the connect path
@@ -288,39 +286,16 @@ def _check_conversation_access(self, user, *, cache_conversation: bool) -> bool:
         if self.conversation_id is None:
             return False
 
-        try:
-            conversation = Conversation.objects.get(pk=self.conversation_id)
-        except Conversation.DoesNotExist:
+        conversation = ConversationService.get_or_none(
+            Conversation, self.conversation_id, user
+        )
+        if conversation is None:
             return False
 
         if cache_conversation:
             self.conversation = conversation
 
-        if conversation.creator_id == user.pk or user.is_superuser:
-            return True
-
-        has_corpus = (
-            Corpus.objects.visible_to_user(user)
-            .filter(pk=conversation.chat_with_corpus_id)
-            .exists()
-            if conversation.chat_with_corpus_id
-            else None
-        )
-        has_document = (
-            Document.objects.visible_to_user(user)
-            .filter(pk=conversation.chat_with_document_id)
-            .exists()
-            if conversation.chat_with_document_id
-            else None
-        )
-
-        if has_corpus is not None and has_document is not None:
-            return has_corpus and has_document
-        if has_corpus is not None:
-            return has_corpus
-        if has_document is not None:
-            return has_document
-        return False
+        return True
 
     # -------------------------------------------------------------------------
     #  AuthHandshakeMixin override
diff --git a/config/websocket/consumers/unified_agent_conversation.py b/config/websocket/consumers/unified_agent_conversation.py
index b15ac1ce09..96a09495b0 100644
--- a/config/websocket/consumers/unified_agent_conversation.py
+++ b/config/websocket/consumers/unified_agent_conversation.py
@@ -65,7 +65,6 @@
     filter_by_scope,
 )
 from opencontractserver.llms.types import AgentFramework
-from opencontractserver.types.enums import PermissionTypes
 from opencontractserver.utils.ids import from_global_id
 
 logger = logging.getLogger(__name__)
@@ -155,7 +154,7 @@ async def connect(self) -> None:
             ):
                 return
 
-            user = self.scope.get("user")
+            user = self.current_user
             is_authenticated = user and user.is_authenticated
 
             if is_authenticated:
@@ -248,36 +247,19 @@ async def _validate_resource_permissions(self, user) -> bool:
         this consumer is currently bound to. Used by AuthHandshakeMixin on
         refresh to detect mid-connection access revocation.
         """
-        is_authenticated = user is not None and user.is_authenticated
+        from opencontractserver.conversations.models import Conversation
+        from opencontractserver.shared.services.base import BaseService
 
-        if self.corpus is not None:
-            if is_authenticated:
-                has_perm = await database_sync_to_async(self.corpus.user_can)(
-                    user, PermissionTypes.READ
-                )
-                if not has_perm:
-                    return False
-            else:
-                # Anonymous fallback: re-fetch is_public from the DB rather
-                # than trusting the in-memory object loaded at connect time.
-                # If the owner flips the corpus to private mid-connection, an
-                # anonymous AUTH refresh would otherwise pass on stale state.
-                fresh_corpus = await Corpus.objects.aget(pk=self.corpus.pk)
-                if not fresh_corpus.is_public:
-                    return False
-
-        if self.document is not None:
-            if is_authenticated:
-                has_perm = await database_sync_to_async(self.document.user_can)(
-                    user, PermissionTypes.READ
+        for model, pk in (
+            (Corpus, self.corpus.pk if self.corpus is not None else None),
+            (Document, self.document.pk if self.document is not None else None),
+            (Conversation, self.conversation_id),
+        ):
+            if pk is not None:
+                visible = await database_sync_to_async(BaseService.get_or_none)(
+                    model, pk, user
                 )
-                if not has_perm:
-                    return False
-            else:
-                # Same anonymous-refresh stale-read concern as the corpus
-                # branch above.
-                fresh_document = await Document.objects.aget(pk=self.document.pk)
-                if not fresh_document.is_public:
+                if visible is None:
                     return False
 
         return True
@@ -349,7 +331,7 @@ async def _resolve_agent_config(self) -> AgentConfiguration | None:
         """
         # Priority 1: Explicit agent_id (visibility-gated)
         if self.agent_config_id:
-            user = self.scope.get("user")
+            user = self.current_user
 
             def _visible_agent_lookup() -> AgentConfiguration | None:
                 return (
@@ -478,7 +460,7 @@ async def receive(self, text_data: str) -> None:
         - Query: {"query": "user question"}
         - Approval: {"approval_decision": true/false, "llm_message_id": 123}
         """
-        logger.debug(f"[Session {self.session_id}] receive(): {text_data[:200]}...")
+        logger.debug("[Session %s] Received WebSocket frame", self.session_id)
 
         try:
             payload: dict[str, Any] = json.loads(text_data)
@@ -936,7 +918,7 @@ async def _resolve_delegation_targets(
         if not slugs:
             return []
 
-        user = self.scope.get("user")
+        user = self.current_user
         corpus_id = self.corpus_id
         document_id = self.document_id
 
diff --git a/config/websocket/middleware.py b/config/websocket/middleware.py
index 637427e2fa..b7c8a5a180 100644
--- a/config/websocket/middleware.py
+++ b/config/websocket/middleware.py
@@ -35,8 +35,10 @@
 """
 
 import logging
+import math
 from typing import Any
 
+import jwt
 from channels.db import database_sync_to_async
 from channels.middleware import BaseMiddleware
 from django.contrib.auth.models import AnonymousUser
@@ -46,6 +48,26 @@
 
 logger = logging.getLogger(__name__)
 
+
+def verified_token_expiry(token: str) -> float:
+    """Read the deadline only AFTER this exact token passed JWT verification.
+
+    Identity and signature checks belong to get_user_from_jwt_token. This
+    second decode only schedules connection expiry; it never authenticates.
+    WebSocket credentials must have a finite expiration.
+    """
+    try:
+        value = jwt.decode(token, options={"verify_signature": False})["exp"]
+        if isinstance(value, bool):
+            raise ValueError("Invalid expiry")
+        expiry = float(value)
+        if not math.isfinite(expiry):
+            raise ValueError("Invalid expiry")
+        return expiry
+    except (KeyError, TypeError, ValueError, jwt.InvalidTokenError) as exc:
+        raise JSONWebTokenError("Invalid token expiry") from exc
+
+
 # Subprotocol marker the client sends and the server echoes.
 # Versioned so we can roll a v2 protocol without breaking existing clients.
 WS_AUTH_SUBPROTOCOL = "opencontracts.jwt.v1"
@@ -131,6 +153,8 @@ async def __call__(self, scope: dict[str, Any], receive: Any, send: Any) -> Any:
         scope["user"] = AnonymousUser()
         scope["auth_error"] = None
         scope["accepted_subprotocol"] = None
+        scope["auth_token"] = None
+        scope["auth_expires_at"] = None
 
         marker_present, token = _parse_subprotocol_token(
             scope.get("headers", []),
@@ -145,7 +169,10 @@ async def __call__(self, scope: dict[str, Any], receive: Any, send: Any) -> Any:
 
         try:
             user = await _get_user_from_token(token)
+            expiry = verified_token_expiry(token)
             scope["user"] = user
+            scope["auth_token"] = token
+            scope["auth_expires_at"] = expiry
             logger.debug(f"WS handshake authenticated user={user.username}")
         except JSONWebTokenExpired as e:
             logger.warning(f"WS handshake auth failed - token expired: {e}")
diff --git a/frontend/src/components/annotator/api/cachedRest.ts b/frontend/src/components/annotator/api/cachedRest.ts
index 20481a07e4..ace870d296 100644
--- a/frontend/src/components/annotator/api/cachedRest.ts
+++ b/frontend/src/components/annotator/api/cachedRest.ts
@@ -11,6 +11,12 @@ import {
 } from "./rest";
 import { DOCX_CACHE_MAX_ENTRIES } from "../../../assets/configurations/constants";
 
+import {
+  assertDocumentCacheGeneration,
+  getDocumentCacheGeneration,
+  docxBytesCache,
+} from "../../../services/documentCacheState";
+
 /**
  * Get PAWLS layer data with caching
  */
@@ -18,6 +24,7 @@ export async function getPawlsLayer(
   url: string,
   documentId?: string
 ): Promise<PageTokens[]> {
+  const generation = getDocumentCacheGeneration();
   console.log(
     `📄 Loading PAWLS data for document ${documentId || "unknown"}...`
   );
@@ -25,6 +32,7 @@ export async function getPawlsLayer(
   // If we have a document ID, try to get from cache first
   if (documentId) {
     const cached = await documentCacheManager.getCachedPawlsData(documentId);
+    assertDocumentCacheGeneration(generation);
     if (cached) {
       console.log(`✅ Loaded PAWLS data from CACHE for document ${documentId}`);
       return cached;
@@ -33,7 +41,9 @@ export async function getPawlsLayer(
 
   // Fetch from server
   console.log(`🌐 Loading PAWLS data from HTTPS: ${url}`);
+  assertDocumentCacheGeneration(generation);
   const pawlsData = await uncachedGetPawlsLayer(url);
+  assertDocumentCacheGeneration(generation);
 
   // Cache for future use if we have document ID
   if (documentId && pawlsData) {
@@ -49,6 +59,7 @@ export async function getPawlsLayer(
       });
   }
 
+  assertDocumentCacheGeneration(generation);
   return pawlsData;
 }
 
@@ -60,11 +71,13 @@ export async function getDocumentRawText(
   documentId?: string,
   hash?: string
 ): Promise<string> {
+  const generation = getDocumentCacheGeneration();
   console.log(`📄 Loading text document ${documentId || "unknown"}...`);
 
   // If we have a document ID, try to get from cache first
   if (documentId) {
     const cached = await documentCacheManager.getCachedText(documentId, hash);
+    assertDocumentCacheGeneration(generation);
     if (cached) {
       console.log(
         `✅ Loaded text document from CACHE for document ${documentId}`
@@ -75,7 +88,9 @@ export async function getDocumentRawText(
 
   // Fetch from server
   console.log(`🌐 Loading text document from HTTPS: ${url}`);
+  assertDocumentCacheGeneration(generation);
   const text = await uncachedGetDocumentRawText(url);
+  assertDocumentCacheGeneration(generation);
 
   // Cache for future use if we have document ID
   if (documentId && text) {
@@ -91,6 +106,7 @@ export async function getDocumentRawText(
       });
   }
 
+  assertDocumentCacheGeneration(generation);
   return text;
 }
 
@@ -103,11 +119,13 @@ export async function getCachedPDFUrl(
   documentId: string,
   hash: string
 ): Promise<string> {
+  const generation = getDocumentCacheGeneration();
   console.log(`📄 Loading PDF document ${documentId}...`);
 
   // First check if we have a valid cached version
   const cachedBlob = await documentCacheManager.getCachedPDF(documentId, hash);
 
+  assertDocumentCacheGeneration(generation);
   if (cachedBlob) {
     console.log(`✅ Loaded PDF from CACHE for document ${documentId}`);
     // Create a blob URL from the cached blob
@@ -130,6 +148,7 @@ export async function getCachedPDFUrl(
       },
     });
 
+    assertDocumentCacheGeneration(generation);
     const pdfBlob = response.data;
 
     // Cache the PDF for future use
@@ -142,11 +161,13 @@ export async function getCachedPDFUrl(
         console.error("  ⚠️ Failed to cache PDF:", err);
       });
 
+    assertDocumentCacheGeneration(generation);
     // Return blob URL for immediate use
     return URL.createObjectURL(pdfBlob);
   } catch (error) {
     console.error("Error fetching PDF:", error);
     // Fall back to direct URL if caching fails
+    assertDocumentCacheGeneration(generation);
     return pdfUrl;
   }
 }
@@ -157,13 +178,13 @@ export async function getCachedPDFUrl(
  * when the limit is reached. Uses Map insertion-order semantics: a cache hit
  * deletes and re-inserts the entry so it becomes the most recent.
  */
-const docxBytesCache = new Map<string, Uint8Array>();
 
 /**
  * Get DOCX document bytes (as Uint8Array) for WASM rendering.
  * Caches by URL to avoid re-downloading on Apollo query refetches.
  */
 export async function getDocxBytes(url: string): Promise<Uint8Array> {
+  const generation = getDocumentCacheGeneration();
   const cached = docxBytesCache.get(url);
   if (cached) {
     // Promote to most-recently-used by re-inserting
@@ -173,6 +194,7 @@ export async function getDocxBytes(url: string): Promise<Uint8Array> {
   }
 
   const response = await axios.get(url, { responseType: "arraybuffer" });
+  assertDocumentCacheGeneration(generation);
   const bytes = new Uint8Array(response.data);
 
   // Evict least-recently-used entry if cache is at capacity
diff --git a/frontend/src/services/__tests__/documentCacheSession.test.ts b/frontend/src/services/__tests__/documentCacheSession.test.ts
new file mode 100644
index 0000000000..032d2d2f94
--- /dev/null
+++ b/frontend/src/services/__tests__/documentCacheSession.test.ts
@@ -0,0 +1,221 @@
+import "fake-indexeddb/auto";
+import FDBFactory from "fake-indexeddb/lib/FDBFactory";
+import axios, { AxiosHeaders } from "axios";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { documentCacheManager } from "../documentCacheManager";
+import {
+  docxBytesCache,
+  getDocumentCacheGeneration,
+  invalidateDocumentCacheSession,
+} from "../documentCacheState";
+import {
+  beginAuthSession,
+  clearAuthSession,
+  authSessionCleanupPendingVar,
+} from "../../utils/authSession";
+import {
+  getDocumentRawText,
+  getPawlsLayer,
+  getCachedPDFUrl,
+  getDocxBytes,
+} from "../../components/annotator/api/cachedRest";
+import * as rest from "../../components/annotator/api/rest";
+
+vi.mock("../../components/annotator/api/rest", () => ({
+  getDocumentRawText: vi.fn(),
+  getPawlsLayer: vi.fn(),
+}));
+
+function deferred<T>() {
+  let resolve!: (value: T) => void;
+  const promise = new Promise<T>((done) => {
+    resolve = done;
+  });
+  return { promise, resolve };
+}
+
+describe("document cache isolation at authentication changes", () => {
+  beforeEach(() => vi.stubGlobal("FDBFactory", FDBFactory));
+
+  afterEach(async () => {
+    vi.restoreAllMocks();
+    vi.unstubAllGlobals();
+    clearAuthSession();
+    await vi.waitFor(() => expect(authSessionCleanupPendingVar()).toBe(false));
+    await documentCacheManager.clearCache();
+  });
+
+  const cacheFormats = [
+    {
+      format: "text",
+      write: () =>
+        documentCacheManager.cacheText("private-doc", "PRIVATE", "hash"),
+      read: () => documentCacheManager.getCachedText("private-doc", "hash"),
+      empty: null,
+    },
+    {
+      format: "pdf",
+      write: () =>
+        documentCacheManager.cachePDF(
+          "private-doc",
+          "hash",
+          new Blob(["%PDF synthetic"], { type: "application/pdf" })
+        ),
+      read: () => documentCacheManager.getCachedPDF("private-doc", "hash"),
+      empty: null,
+    },
+    {
+      format: "pawls",
+      write: () => documentCacheManager.cachePawlsData("private-doc", []),
+      read: () => documentCacheManager.getCachedPawlsData("private-doc"),
+      empty: null,
+    },
+  ];
+  const cacheReads = [
+    ...cacheFormats,
+    {
+      format: "metadata",
+      write: cacheFormats[0].write,
+      read: () =>
+        documentCacheManager.validateCache("private-doc", "text", "hash"),
+      empty: false,
+    },
+  ];
+
+  it.each(cacheReads)(
+    "discards $format reads when the session changes while opening storage",
+    async ({ write, read, empty }) => {
+      await write();
+      expect(await read()).not.toBe(empty);
+      const pending = read();
+      invalidateDocumentCacheSession();
+      expect(await pending).toBe(empty);
+    }
+  );
+
+  it.each(cacheReads)(
+    "discards $format reads when the session changes before storage responds",
+    async ({ write, read, empty }) => {
+      await write();
+      const get = IDBObjectStore.prototype.get;
+      vi.spyOn(IDBObjectStore.prototype, "get").mockImplementationOnce(
+        function (this: IDBObjectStore, key: IDBValidKey | IDBKeyRange) {
+          const request = get.call(this, key);
+          // The request still reads the old entry from real fake-indexeddb storage.
+          // Only the authentication session changes before its callback runs.
+          invalidateDocumentCacheSession();
+          return request;
+        }
+      );
+      expect(await read()).toBe(empty);
+    }
+  );
+
+  it.each(cacheFormats)(
+    "discards $format writes when the session changes while opening storage",
+    async ({ write }) => {
+      const pending = write();
+      invalidateDocumentCacheSession();
+      await pending;
+      expect((await documentCacheManager.getCacheStats()).count).toBe(0);
+    }
+  );
+
+  it.each(cacheFormats)(
+    "discards $format writes when the session changes while checking capacity",
+    async ({ write }) => {
+      const openCursor = IDBObjectStore.prototype.openCursor;
+      vi.spyOn(IDBObjectStore.prototype, "openCursor").mockImplementationOnce(
+        function (
+          this: IDBObjectStore,
+          ...args: Parameters<IDBObjectStore["openCursor"]>
+        ) {
+          const request = openCursor.apply(this, args);
+          invalidateDocumentCacheSession();
+          return request;
+        }
+      );
+      await write();
+      expect((await documentCacheManager.getCacheStats()).count).toBe(0);
+    }
+  );
+
+  it("invalidates memory caches even when IndexedDB is unavailable", async () => {
+    const generation = getDocumentCacheGeneration();
+    docxBytesCache.set("private.docx", new Uint8Array([1, 2]));
+    vi.stubGlobal("indexedDB", undefined);
+    await documentCacheManager.clearCache();
+    expect(getDocumentCacheGeneration()).toBe(generation + 1);
+    expect(docxBytesCache.size).toBe(0);
+  });
+
+  it("clears stored content and DOCX bytes on logout/account replacement", async () => {
+    beginAuthSession("synthetic-owner-token");
+    await documentCacheManager.cacheText("private-doc", "OWNER_PRIVATE_TEXT");
+    expect(await documentCacheManager.getCachedText("private-doc")).toBe(
+      "OWNER_PRIVATE_TEXT"
+    );
+    docxBytesCache.set("private.docx", new Uint8Array([1, 2]));
+    clearAuthSession(undefined, "logout");
+    expect(docxBytesCache.size).toBe(0);
+    expect(await documentCacheManager.getCachedText("private-doc")).toBeNull();
+    await vi.waitFor(() => expect(authSessionCleanupPendingVar()).toBe(false));
+    beginAuthSession("synthetic-stranger-token");
+    expect(await documentCacheManager.getCachedText("private-doc")).toBeNull();
+  });
+
+  it("isolates old storage even when deletion fails", async () => {
+    await documentCacheManager.cacheText("private-doc", "OWNER_PRIVATE_TEXT");
+    // Changing the namespace is synchronous; safety never depends on IDB deletion.
+    invalidateDocumentCacheSession();
+    expect(await documentCacheManager.getCachedText("private-doc")).toBeNull();
+  });
+
+  it.each(["text", "pawls", "pdf", "docx"])(
+    "rejects late %s responses after logout",
+    async (format) => {
+      const network = deferred<void>();
+      let pending: Promise<unknown>;
+      if (format === "text") {
+        vi.mocked(rest.getDocumentRawText).mockReturnValueOnce(
+          network.promise.then(() => "PRIVATE")
+        );
+        pending = getDocumentRawText("private.txt", "private-doc");
+        await vi.waitFor(() =>
+          expect(rest.getDocumentRawText).toHaveBeenCalled()
+        );
+      } else if (format === "pawls") {
+        vi.mocked(rest.getPawlsLayer).mockReturnValueOnce(
+          network.promise.then(() => [])
+        );
+        pending = getPawlsLayer("private.json", "private-doc");
+        await vi.waitFor(() => expect(rest.getPawlsLayer).toHaveBeenCalled());
+      } else {
+        vi.spyOn(axios, "get").mockReturnValueOnce(
+          network.promise.then(() => ({
+            data: new Uint8Array([1, 2]).buffer,
+            status: 200,
+            statusText: "OK",
+            headers: {},
+            config: { headers: new AxiosHeaders() },
+          }))
+        );
+        pending =
+          format === "pdf"
+            ? getCachedPDFUrl("private.pdf", "private-doc", "hash")
+            : getDocxBytes("private.docx");
+        await vi.waitFor(() => expect(axios.get).toHaveBeenCalled());
+      }
+      const rejected = expect(pending).rejects.toThrow(
+        "Document session changed"
+      );
+      clearAuthSession(undefined, "logout");
+      network.resolve();
+      await rejected;
+      expect(
+        await documentCacheManager.getCachedText("private-doc")
+      ).toBeNull();
+      expect(docxBytesCache.size).toBe(0);
+    }
+  );
+});
diff --git a/frontend/src/services/documentCacheManager.ts b/frontend/src/services/documentCacheManager.ts
index bdee43f258..fbfc211d1d 100644
--- a/frontend/src/services/documentCacheManager.ts
+++ b/frontend/src/services/documentCacheManager.ts
@@ -5,6 +5,12 @@
  * with hash-based validation. Implements LRU eviction strategy and respects storage limits.
  */
 
+import {
+  getDocumentCacheGeneration,
+  getDocumentCacheScope,
+  invalidateDocumentCacheSession,
+} from "./documentCacheState";
+
 interface CachedDocument {
   documentId: string;
   documentType: "pdf" | "text" | "pawls";
@@ -85,7 +91,7 @@ export class DocumentCacheManager {
     documentId: string,
     documentType: "pdf" | "text" | "pawls"
   ): string {
-    return `${documentId}_${documentType}`;
+    return `${getDocumentCacheScope()}:${documentId}_${documentType}`;
   }
 
   /**
@@ -95,8 +101,10 @@ export class DocumentCacheManager {
     documentId: string,
     expectedHash: string
   ): Promise<Blob | null> {
+    const generation = getDocumentCacheGeneration();
     try {
       const db = await this.initDB();
+      if (generation !== getDocumentCacheGeneration()) return null;
       const transaction = db.transaction([this.STORE_NAME], "readonly");
       const store = transaction.objectStore(this.STORE_NAME);
       const cacheKey = this.getCacheKey(documentId, "pdf");
@@ -106,6 +114,10 @@ export class DocumentCacheManager {
 
         request.onsuccess = () => {
           const cached = request.result as CachedDocument | undefined;
+          if (generation !== getDocumentCacheGeneration()) {
+            resolve(null);
+            return;
+          }
 
           if (!cached) {
             console.log(`No cached PDF found for document ${documentId}`);
@@ -174,8 +186,10 @@ export class DocumentCacheManager {
     documentId: string,
     expectedHash?: string
   ): Promise<string | null> {
+    const generation = getDocumentCacheGeneration();
     try {
       const db = await this.initDB();
+      if (generation !== getDocumentCacheGeneration()) return null;
       const transaction = db.transaction([this.STORE_NAME], "readonly");
       const store = transaction.objectStore(this.STORE_NAME);
       const cacheKey = this.getCacheKey(documentId, "text");
@@ -185,6 +199,10 @@ export class DocumentCacheManager {
 
         request.onsuccess = async () => {
           const cached = request.result as CachedDocument | undefined;
+          if (generation !== getDocumentCacheGeneration()) {
+            resolve(null);
+            return;
+          }
 
           if (!cached) {
             console.log(`No cached text found for document ${documentId}`);
@@ -222,7 +240,9 @@ export class DocumentCacheManager {
             } else if (cached.data instanceof Blob) {
               // Legacy format - stored as Blob (for backward compatibility)
               const text = await cached.data.text();
-              resolve(text);
+              resolve(
+                generation === getDocumentCacheGeneration() ? text : null
+              );
             } else if (
               cached.data &&
               typeof cached.data === "object" &&
@@ -259,8 +279,10 @@ export class DocumentCacheManager {
    * Get cached PAWLS data if it exists
    */
   async getCachedPawlsData(documentId: string): Promise<any | null> {
+    const generation = getDocumentCacheGeneration();
     try {
       const db = await this.initDB();
+      if (generation !== getDocumentCacheGeneration()) return null;
       const transaction = db.transaction([this.STORE_NAME], "readonly");
       const store = transaction.objectStore(this.STORE_NAME);
       const cacheKey = this.getCacheKey(documentId, "pawls");
@@ -270,6 +292,10 @@ export class DocumentCacheManager {
 
         request.onsuccess = () => {
           const cached = request.result as CachedDocument | undefined;
+          if (generation !== getDocumentCacheGeneration()) {
+            resolve(null);
+            return;
+          }
 
           if (!cached) {
             console.log(
@@ -309,8 +335,10 @@ export class DocumentCacheManager {
    * Cache a PDF with its hash
    */
   async cachePDF(documentId: string, hash: string, blob: Blob): Promise<void> {
+    const generation = getDocumentCacheGeneration();
     try {
       const db = await this.initDB();
+      if (generation !== getDocumentCacheGeneration()) return;
 
       // Check current cache size and evict if necessary
       const currentSize = await this.getCacheSize();
@@ -338,6 +366,7 @@ export class DocumentCacheManager {
         });
 
         // Now start the transaction with the data ready
+        if (generation !== getDocumentCacheGeneration()) return;
         const transaction = db.transaction(
           [this.STORE_NAME, this.METADATA_STORE_NAME],
           "readwrite"
@@ -382,6 +411,7 @@ export class DocumentCacheManager {
         });
       } else {
         // Real browser environment - store Blob normally
+        if (generation !== getDocumentCacheGeneration()) return;
         const transaction = db.transaction(
           [this.STORE_NAME, this.METADATA_STORE_NAME],
           "readwrite"
@@ -437,9 +467,11 @@ export class DocumentCacheManager {
     text: string,
     hash?: string
   ): Promise<void> {
+    const generation = getDocumentCacheGeneration();
     try {
       // Store text directly as string instead of Blob for better compatibility
       const db = await this.initDB();
+      if (generation !== getDocumentCacheGeneration()) return;
 
       // Check current cache size and evict if necessary
       const currentSize = await this.getCacheSize();
@@ -449,6 +481,7 @@ export class DocumentCacheManager {
         await this.evictOldest(newSize);
       }
 
+      if (generation !== getDocumentCacheGeneration()) return;
       const transaction = db.transaction(
         [this.STORE_NAME, this.METADATA_STORE_NAME],
         "readwrite"
@@ -501,10 +534,12 @@ export class DocumentCacheManager {
    * Cache PAWLS data
    */
   async cachePawlsData(documentId: string, pawlsData: any): Promise<void> {
+    const generation = getDocumentCacheGeneration();
     try {
       const dataStr = JSON.stringify(pawlsData);
       const size = new Blob([dataStr]).size;
       const db = await this.initDB();
+      if (generation !== getDocumentCacheGeneration()) return;
 
       // Check current cache size and evict if necessary
       const currentSize = await this.getCacheSize();
@@ -513,6 +548,7 @@ export class DocumentCacheManager {
         await this.evictOldest(size);
       }
 
+      if (generation !== getDocumentCacheGeneration()) return;
       const transaction = db.transaction(
         [this.STORE_NAME, this.METADATA_STORE_NAME],
         "readwrite"
@@ -567,8 +603,10 @@ export class DocumentCacheManager {
     documentType: "pdf" | "text" | "pawls",
     serverHash?: string
   ): Promise<boolean> {
+    const generation = getDocumentCacheGeneration();
     try {
       const db = await this.initDB();
+      if (generation !== getDocumentCacheGeneration()) return false;
       const transaction = db.transaction(
         [this.METADATA_STORE_NAME],
         "readonly"
@@ -581,6 +619,10 @@ export class DocumentCacheManager {
 
         request.onsuccess = () => {
           const metadata = request.result as CacheMetadata | undefined;
+          if (generation !== getDocumentCacheGeneration()) {
+            resolve(false);
+            return;
+          }
 
           if (!metadata) {
             resolve(false);
@@ -773,6 +815,10 @@ export class DocumentCacheManager {
    * Clear all cached documents
    */
   async clearCache(): Promise<void> {
+    // Invalidate synchronously, including in-flight reads/writes and DOCX bytes.
+    // A new scope stays isolated even if IndexedDB cleanup fails.
+    invalidateDocumentCacheSession();
+    if (typeof indexedDB === "undefined") return;
     try {
       const db = await this.initDB();
       const transaction = db.transaction(
diff --git a/frontend/src/services/documentCacheState.ts b/frontend/src/services/documentCacheState.ts
new file mode 100644
index 0000000000..e964678c70
--- /dev/null
+++ b/frontend/src/services/documentCacheState.ts
@@ -0,0 +1,16 @@
+/** Document bytes belong to one page session and authentication generation. */
+const pageScope = crypto.randomUUID();
+let generation = 0;
+
+export const getDocumentCacheGeneration = () => generation;
+export const getDocumentCacheScope = () => `${pageScope}:${generation}`;
+export const docxBytesCache = new Map<string, Uint8Array>();
+
+export function invalidateDocumentCacheSession(): void {
+  generation++;
+  docxBytesCache.clear();
+}
+
+export function assertDocumentCacheGeneration(expected: number): void {
+  if (generation !== expected) throw new Error("Document session changed");
+}
diff --git a/frontend/src/utils/__tests__/docxodusWasm.test.ts b/frontend/src/utils/__tests__/docxodusWasm.test.ts
new file mode 100644
index 0000000000..b94585cd9c
--- /dev/null
+++ b/frontend/src/utils/__tests__/docxodusWasm.test.ts
@@ -0,0 +1,75 @@
+// @vitest-environment node
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import fs from "node:fs";
+import os from "node:os";
+import path from "node:path";
+import { docxodusWasmPlugin } from "../../../tooling/docxodusWasm";
+
+describe("WASM asset containment", () => {
+  let directory: string;
+  let middleware: Function;
+  beforeEach(() => {
+    directory = fs.mkdtempSync(path.join(os.tmpdir(), "oc-wasm-security-"));
+    const root = path.join(directory, "wasm");
+    fs.mkdirSync(root);
+    fs.writeFileSync(path.join(directory, "private.json"), "synthetic secret");
+    fs.writeFileSync(path.join(root, "runtime.wasm"), "wasm asset");
+    fs.writeFileSync(path.join(root, "runtimeconfig.bin"), "runtime config");
+    fs.symlinkSync(
+      path.join(directory, "private.json"),
+      path.join(root, "link.json")
+    );
+    const plugin = docxodusWasmPlugin(root);
+    (plugin.configureServer as Function)({
+      middlewares: {
+        use: (handler: Function) => {
+          middleware = handler;
+        },
+      },
+    });
+  });
+  afterEach(() => fs.rmSync(directory, { recursive: true, force: true }));
+  function request(suffix: string) {
+    const response = { setHeader: vi.fn(), writeHead: vi.fn(), end: vi.fn() };
+    const next = vi.fn();
+    middleware(
+      { url: `/node_modules/docxodus/dist/wasm/${suffix}` },
+      response,
+      next
+    );
+    return { response, next };
+  }
+  it.each(["../private.json", "%2e%2e%2fprivate.json", "link.json", "%ZZ"])(
+    "rejects %s",
+    (suffix) => {
+      const { response } = request(suffix);
+      expect(response.writeHead).toHaveBeenCalledWith(403);
+      expect(response.end).toHaveBeenCalledWith();
+    }
+  );
+  it("never treats query-string traversal as a filesystem path", () => {
+    const { response } = request("runtime.wasm?x/../../private.json");
+    expect(response.end).toHaveBeenCalledWith(Buffer.from("wasm asset"));
+  });
+  it("serves the runtime configuration binary", () => {
+    const { response } = request("runtimeconfig.bin");
+    expect(response.end).toHaveBeenCalledWith(Buffer.from("runtime config"));
+    expect(response.setHeader).toHaveBeenCalledWith(
+      "Content-Type",
+      "application/octet-stream"
+    );
+  });
+  it("serves legitimate assets with query strings without wildcard CORS", () => {
+    const { response, next } = request("runtime.wasm?v=1");
+    expect(response.end).toHaveBeenCalledWith(Buffer.from("wasm asset"));
+    expect(response.setHeader).toHaveBeenCalledWith(
+      "Content-Type",
+      "application/wasm"
+    );
+    expect(response.setHeader).not.toHaveBeenCalledWith(
+      "Access-Control-Allow-Origin",
+      "*"
+    );
+    expect(next).not.toHaveBeenCalled();
+  });
+});
diff --git a/frontend/src/utils/authSession.ts b/frontend/src/utils/authSession.ts
index 7a1292f6a5..9531df6503 100644
--- a/frontend/src/utils/authSession.ts
+++ b/frontend/src/utils/authSession.ts
@@ -10,6 +10,7 @@ import {
 } from "../graphql/cache";
 import { clearLocalAuthSession } from "./localAuthSession";
 import { makeVar } from "@apollo/client";
+import { documentCacheManager } from "../services/documentCacheManager";
 
 export const authSessionEpochVar = makeVar(0);
 export const authSessionCleanupPendingVar = makeVar(false);
@@ -20,6 +21,9 @@ export const getAuthSessionEpoch = () => authSessionEpochVar();
 
 /** A new login is distinct from the SDK renewing the same user's token. */
 export function beginAuthSession(token: string): void {
+  void documentCacheManager.clearCache().catch(() => {
+    console.warn("Unable to remove stored document content");
+  });
   authSessionEpochVar(authSessionEpochVar() + 1);
   authToken(token);
 }
@@ -52,6 +56,7 @@ export function replaceAuthSession(expectedToken: string): void {
 }
 
 function resetAuthSession(token: string, reason: ClearReason): void {
+  const documentCleanup = documentCacheManager.clearCache();
   // Close the routing gate before advancing the epoch or clearing the store.
   // Overlapping invalidations must all settle before requests can restart.
   authInitCompleteVar(false);
@@ -67,9 +72,10 @@ function resetAuthSession(token: string, reason: ClearReason): void {
   editingDocument(null);
   showUserSettingsModal(false);
   cache.restore({});
-  void Promise.allSettled(
-    Array.from(cleanupHandlers, async (cleanup) => cleanup(reason))
-  ).then((results) => {
+  void Promise.allSettled([
+    documentCleanup,
+    ...Array.from(cleanupHandlers, async (cleanup) => cleanup(reason)),
+  ]).then((results) => {
     if (results.some((result) => result.status === "rejected")) {
       console.warn("Unable to finish clearing the authentication cache");
     }
diff --git a/frontend/tooling/docxodusWasm.ts b/frontend/tooling/docxodusWasm.ts
new file mode 100644
index 0000000000..40102ac22f
--- /dev/null
+++ b/frontend/tooling/docxodusWasm.ts
@@ -0,0 +1,54 @@
+import fs from "node:fs";
+import path from "node:path";
+import type { Plugin } from "vite";
+
+const MIME_TYPES: Record<string, string> = {
+  ".js": "application/javascript",
+  ".wasm": "application/wasm",
+  ".json": "application/json",
+  ".dat": "application/octet-stream",
+};
+
+function isContained(root: string, candidate: string): boolean {
+  const relative = path.relative(root, candidate);
+  return (
+    relative !== ".." &&
+    !relative.startsWith(`..${path.sep}`) &&
+    !path.isAbsolute(relative)
+  );
+}
+
+/** Serve only packaged WASM assets; custom middleware bypasses Vite's fs guard. */
+export function docxodusWasmPlugin(assetRoot: string): Plugin {
+  const prefix = "/node_modules/docxodus/dist/wasm/";
+  return {
+    name: "docxodus-wasm-server",
+    configureServer(server) {
+      server.middlewares.use((req, res, next) => {
+        const pathname = (req.url || "").split("?")[0];
+        if (!pathname.startsWith(prefix)) return next();
+        const deny = () => {
+          res.writeHead(403);
+          res.end();
+        };
+        try {
+          const suffix = decodeURIComponent(pathname.slice(prefix.length));
+          const root = fs.realpathSync(assetRoot);
+          const candidate = path.resolve(root, suffix);
+          if (!isContained(root, candidate)) return deny();
+          // Resolve symlinks too: a lexical prefix check alone is insufficient.
+          const filePath = fs.realpathSync(candidate);
+          if (!isContained(root, filePath)) return deny();
+          const mimeType =
+            MIME_TYPES[path.extname(filePath)] || "application/octet-stream";
+          const data = fs.readFileSync(filePath);
+          res.setHeader("Content-Type", mimeType);
+          res.end(data);
+        } catch (error) {
+          if (error instanceof URIError) return deny();
+          next();
+        }
+      });
+    },
+  };
+}
diff --git a/frontend/vite.config.ts b/frontend/vite.config.ts
index 12767ce646..93081e8ebf 100644
--- a/frontend/vite.config.ts
+++ b/frontend/vite.config.ts
@@ -1,6 +1,6 @@
 import { defineConfig } from "vitest/config";
 import react from "@vitejs/plugin-react-swc";
-import fs from "fs";
+import { docxodusWasmPlugin } from "./tooling/docxodusWasm";
 import path from "path";
 
 // Custom plugin to handle asset imports in Playwright tests
@@ -16,59 +16,6 @@ const assetPlugin = () => {
   };
 };
 
-// Serve docxodus WASM files from node_modules with correct MIME types.
-// Vite's dep optimizer rewrites import.meta.url, breaking auto-detection
-// of sibling WASM files. We exclude docxodus from optimization (below)
-// so import.meta.url resolves to the real node_modules path, then this
-// middleware serves the _framework files that the .NET WASM loader fetches.
-const docxodusWasmPlugin = () => {
-  const MIME_TYPES: Record<string, string> = {
-    ".js": "application/javascript",
-    ".wasm": "application/wasm",
-    ".json": "application/json",
-    ".dat": "application/octet-stream",
-  };
-
-  return {
-    name: "docxodus-wasm-server",
-    configureServer(server: { middlewares: { use: Function } }) {
-      server.middlewares.use(
-        (
-          req: { url?: string },
-          res: {
-            setHeader: Function;
-            writeHead: Function;
-            end: Function;
-          },
-          next: Function
-        ) => {
-          const url = req.url || "";
-          // Match requests for docxodus WASM framework files
-          const match = url.match(/\/node_modules\/docxodus\/dist\/wasm\/(.*)/);
-          if (!match) return next();
-
-          const filePath = path.join(
-            __dirname,
-            "node_modules/docxodus/dist/wasm",
-            match[1]
-          );
-          const ext = path.extname(filePath);
-          const mimeType = MIME_TYPES[ext] || "application/octet-stream";
-
-          try {
-            const data = fs.readFileSync(filePath);
-            res.setHeader("Content-Type", mimeType);
-            res.setHeader("Access-Control-Allow-Origin", "*");
-            res.end(data);
-          } catch {
-            next();
-          }
-        }
-      );
-    },
-  };
-};
-
 /**
  * Load the Istanbul instrumentation plugin via dynamic import().
  * vite-plugin-istanbul v8 is ESM-only (exports.require is null),
@@ -110,7 +57,9 @@ export default defineConfig(async () => {
     plugins: [
       react(),
       assetPlugin(),
-      docxodusWasmPlugin(),
+      docxodusWasmPlugin(
+        path.join(__dirname, "node_modules/docxodus/dist/wasm")
+      ),
       // Instrument source code with Istanbul when collecting Playwright CT coverage
       ...istanbulPlugins,
     ],
diff --git a/opencontractserver/document_imports/services.py b/opencontractserver/document_imports/services.py
index 3ed9a53317..c4b8ab8566 100644
--- a/opencontractserver/document_imports/services.py
+++ b/opencontractserver/document_imports/services.py
@@ -1000,7 +1000,7 @@ def start_chunked_upload(
     Raises :class:`ChunkedUploadError` (client error, carries HTTP status)
     or :class:`DocumentImportPermissionError` (permission, 403).
     """
-    metadata = metadata or {}
+    metadata = dict(metadata or {})
 
     if kind not in ChunkedUploadKind.values:
         raise ChunkedUploadError(f"Unknown upload kind: {kind}")
@@ -1032,6 +1032,12 @@ def start_chunked_upload(
             http_status=403,
         )
 
+    # Persist the token's implicit target now. Completion with another token
+    # must not retarget a DOCUMENT upload that omitted add_to_corpus_id.
+    if access_token is not None and kind == ChunkedUploadKind.DOCUMENT:
+        if normalise_optional(metadata.get("add_to_corpus_id")) is None:
+            metadata["add_to_corpus_id"] = str(access_token.corpus_id)
+
     # --- per-kind fast-fail permission gates (see _gate_chunked_corpus) ----------
     if kind == ChunkedUploadKind.DOCUMENT:
         if access_token is None:
@@ -1095,7 +1101,7 @@ def start_chunked_upload(
     return session
 
 
-def _get_owned_session(user, upload_id) -> ChunkedUploadSession:
+def _get_owned_session(user, upload_id, access_token=None) -> ChunkedUploadSession:
     """
     Fetch a session the requester owns, or raise a generic 404.
 
@@ -1103,7 +1109,23 @@ def _get_owned_session(user, upload_id) -> ChunkedUploadSession:
     the IDOR: a cross-user id is indistinguishable from a missing one.
     """
     try:
-        return ChunkedUploadSession.objects.get(id=upload_id, creator=user)
+        session = ChunkedUploadSession.objects.get(id=upload_id, creator=user)
+        if access_token is not None:
+            target_fields: dict[str, str] = {
+                ChunkedUploadKind.DOCUMENT: "add_to_corpus_id",
+                ChunkedUploadKind.ZIP_TO_CORPUS: "corpus_id",
+            }
+            target_field = target_fields.get(session.kind)
+            target = (
+                (session.metadata or {}).get(target_field) if target_field else None
+            )
+            # Legacy unbound worker sessions cannot be safely attributed to
+            # a corpus and must be restarted. Do not mutate them on denial.
+            if target is None or str(_resolve_pk(target)) != str(
+                access_token.corpus_id
+            ):
+                raise ChunkedUploadSession.DoesNotExist
+        return session
     except (ChunkedUploadSession.DoesNotExist, ValueError, TypeError):
         raise ChunkedUploadError("Upload session not found", http_status=404)
 
@@ -1114,6 +1136,7 @@ def store_chunk(
     upload_id,
     index: int,
     chunk_file: UploadedFile,
+    access_token: CorpusAccessToken | None = None,
 ) -> ChunkedSessionInfo:
     """
     Persist one part of a chunked upload (idempotent on ``index``).
@@ -1121,7 +1144,7 @@ def store_chunk(
     Re-uploading an index overwrites the previous part (deleting its
     storage object first) so a client can safely retry a failed part.
     """
-    session = _get_owned_session(user, upload_id)
+    session = _get_owned_session(user, upload_id, access_token)
     if session.status != ChunkedUploadStatus.PENDING:
         raise ChunkedUploadError(
             "Upload session is not accepting parts", http_status=409
@@ -1167,9 +1190,11 @@ def store_chunk(
     return _session_info(locked)
 
 
-def get_chunked_session_status(*, user, upload_id) -> ChunkedSessionInfo:
+def get_chunked_session_status(
+    *, user, upload_id, access_token: CorpusAccessToken | None = None
+) -> ChunkedSessionInfo:
     """Return progress for a session the requester owns (resumability)."""
-    return _session_info(_get_owned_session(user, upload_id))
+    return _session_info(_get_owned_session(user, upload_id, access_token))
 
 
 def _safe_unlink(path: str) -> None:
@@ -1251,7 +1276,7 @@ def complete_chunked_upload(
     :class:`DocumentImportPermissionError` (propagated from the import
     service).
     """
-    session = _get_owned_session(user, upload_id)
+    session = _get_owned_session(user, upload_id, access_token)
     if session.status != ChunkedUploadStatus.PENDING:
         raise ChunkedUploadError(
             f"Upload session is not completable (status={session.status})",
diff --git a/opencontractserver/document_imports/views.py b/opencontractserver/document_imports/views.py
index 7b4c15b12d..7fdd7f77dd 100644
--- a/opencontractserver/document_imports/views.py
+++ b/opencontractserver/document_imports/views.py
@@ -499,6 +499,7 @@ def put(self, request: Request, upload_id: str, index: int) -> Response:
                 upload_id=upload_id,
                 index=index,
                 chunk_file=chunk_file,
+                access_token=_request_access_token(request),
             )
         except ChunkedUploadError as e:
             return _chunked_error_response(e)
@@ -556,7 +557,11 @@ class ChunkedUploadStatusView(APIView):
 
     def get(self, request: Request, upload_id: str) -> Response:
         try:
-            info = get_chunked_session_status(user=request.user, upload_id=upload_id)
+            info = get_chunked_session_status(
+                user=request.user,
+                upload_id=upload_id,
+                access_token=_request_access_token(request),
+            )
         except ChunkedUploadError as e:
             return _chunked_error_response(e)
 
diff --git a/opencontractserver/llms/tools/pydantic_ai_tools.py b/opencontractserver/llms/tools/pydantic_ai_tools.py
index 88433ff074..0e697559e5 100644
--- a/opencontractserver/llms/tools/pydantic_ai_tools.py
+++ b/opencontractserver/llms/tools/pydantic_ai_tools.py
@@ -125,6 +125,9 @@ async def _check_user_permissions(
     except User.DoesNotExist:
         raise PermissionError(f"User {user_id} not found")
 
+    if not user.is_active:
+        raise PermissionError("User is inactive")
+
     if document_id:
         # Use visible_to_user() queryset which properly handles creator
         # access, public status, and guardian permissions.
diff --git a/opencontractserver/shared/services/tree_traversal.py b/opencontractserver/shared/services/tree_traversal.py
new file mode 100644
index 0000000000..d1220adda9
--- /dev/null
+++ b/opencontractserver/shared/services/tree_traversal.py
@@ -0,0 +1,63 @@
+"""Permission-filtered traversal for annotation and note trees."""
+
+from typing import Any, Literal
+
+from django_cte import CTE, with_cte
+
+from opencontractserver.shared.services.base import BaseService
+
+
+class TreeTraversalService(BaseService):
+    @classmethod
+    def get_nodes(
+        cls,
+        root: Any,
+        user: Any,
+        *,
+        mode: Literal["descendants", "full", "subtree"],
+        text_field: str,
+        request: Any = None,
+    ) -> list[dict[str, Any]]:
+        """Traverse only visible edges, stopping at inaccessible ancestors.
+
+        Filter both terms of the recursive query: a readable root does not
+        imply that its children share its analysis/extract privacy. UNION
+        deduplication and the ancestor visited set also terminate cycles.
+        """
+        visible = cls.filter_visible(type(root), user, request=request).order_by()
+        fields = ("id", "parent_id", text_field)
+        current = visible.filter(pk=root.pk).values(*fields).first()
+        if current is None:
+            return []
+
+        ancestors = {current["id"]: current}
+        if mode != "descendants":
+            while current["parent_id"] is not None:
+                parent_id = current["parent_id"]
+                if parent_id in ancestors:
+                    break
+                parent = visible.filter(pk=parent_id).values(*fields).first()
+                if parent is None:
+                    break
+                ancestors[parent_id] = parent
+                current = parent
+
+        def descendants(cte):
+            seed = (
+                visible.filter(pk=current["id"])
+                if mode == "full"
+                else visible.filter(parent_id=root.pk)
+            )
+            children = cte.join(visible, parent_id=cte.col.id)
+            return seed.values(*fields).union(children.values(*fields))
+
+        cte = CTE.recursive(descendants)
+        rows = {
+            row["id"]: row
+            for row in with_cte(cte, select=cte.queryset()).order_by("id")
+        }
+        if mode == "subtree":
+            rows.update(ancestors)
+        elif mode == "descendants":
+            rows.pop(root.pk, None)
+        return [rows[pk] for pk in sorted(rows)]
diff --git a/opencontractserver/tests/architecture/test_graphql_service_layer.py b/opencontractserver/tests/architecture/test_graphql_service_layer.py
index 4914004f0f..1a10fbac10 100644
--- a/opencontractserver/tests/architecture/test_graphql_service_layer.py
+++ b/opencontractserver/tests/architecture/test_graphql_service_layer.py
@@ -371,3 +371,79 @@ def test_django_check_returns_empty_when_repo_is_clean() -> None:
     from opencontractserver.shared.checks import check_graphql_service_layer
 
     assert check_graphql_service_layer(app_configs=None) == []
+
+
+def test_nullable_resource_links_do_not_use_default_fk_resolution():
+    """Default Strawberry getattr bypasses target visibility hooks.
+
+    These resources have independent ACLs. Intrinsic objects such as an
+    annotation's label deliberately inherit their parent's presentation.
+    Non-null relations and non-model result DTOs need separate review.
+    """
+    import ast
+    from pathlib import Path
+
+    independent_types = {
+        "AnalysisType",
+        "ExtractType",
+        "AnalyzerType",
+        "AnnotationType",
+        "MessageType",
+        "CorpusType",
+        "DocumentType",
+        "FieldsetType",
+        "AgentConfigurationType",
+        "GremlinEngineType_WRITE",
+        "LabelSetType",
+        "NoteType",
+    }
+    graphql_dir = Path(__file__).resolve().parents[3] / "config" / "graphql"
+    violations = []
+    for path in graphql_dir.glob("*_types.py"):
+        for node in ast.walk(ast.parse(path.read_text())):
+            if not isinstance(node, ast.AnnAssign) or node.value is None:
+                continue
+            if isinstance(node.value, ast.Call) and any(
+                keyword.arg == "resolver" for keyword in node.value.keywords
+            ):
+                continue
+            annotation = ast.unparse(node.annotation)
+            if "None" not in annotation or "list[" in annotation:
+                continue
+            names = {
+                part.id
+                for part in ast.walk(node.annotation)
+                if isinstance(part, ast.Name)
+            }
+            if names & independent_types:
+                violations.append(
+                    f"{path.name}:{node.lineno}: {ast.unparse(node.target)}"
+                )
+    assert (
+        not violations
+    ), "Use resolve_visible_fk for independently private resources: " + ", ".join(
+        violations
+    )
+
+
+def test_independent_fk_targets_have_permission_hooks():
+    from config.graphql.core.relay import _TYPE_REGISTRY
+    from config.graphql.schema import schema
+
+    assert schema is not None  # Populate the registry using the served schema.
+    for name in (
+        "AnalysisType",
+        "ExtractType",
+        "AnalyzerType",
+        "AnnotationType",
+        "MessageType",
+        "CorpusType",
+        "DocumentType",
+        "FieldsetType",
+        "AgentConfigurationType",
+        "GremlinEngineType_WRITE",
+        "LabelSetType",
+        "NoteType",
+    ):
+        entry = _TYPE_REGISTRY[name]
+        assert entry.get_node is not None or entry.get_queryset is not None, name
diff --git a/opencontractserver/tests/test_authentication_logging.py b/opencontractserver/tests/test_authentication_logging.py
new file mode 100644
index 0000000000..699e99f8e1
--- /dev/null
+++ b/opencontractserver/tests/test_authentication_logging.py
@@ -0,0 +1,38 @@
+"""Authentication diagnostic logging contracts."""
+
+from unittest.mock import patch
+
+from django.test import TestCase
+
+
+class AuthenticationLoggingTests(TestCase):
+    def test_api_token_is_absent_from_debug_logs(self):
+        from django.test import RequestFactory, override_settings
+
+        from config.graphql_api_token_auth.backends import ApiKeyBackend
+
+        secret = "SYNTHETIC_API_TOKEN_MUST_NOT_APPEAR"
+        request = RequestFactory().get(
+            "/graphql/", HTTP_AUTHORIZATION=f"Api-Token {secret}"
+        )
+        with override_settings(API_TOKEN_PREFIX="Api-Token"), patch.object(
+            ApiKeyBackend, "authenticate_credentials"
+        ), self.assertLogs(
+            "config.graphql_api_token_auth.backends", level="DEBUG"
+        ) as logs:
+            ApiKeyBackend().authenticate(request)
+        self.assertNotIn(secret, "\n".join(logs.output))
+
+    def test_auth0_backend_does_not_log_password_kwargs(self):
+        from config.graphql_auth0_auth.backends import (
+            Auth0RemoteUserJSONWebTokenBackend,
+        )
+
+        secret = "SYNTHETIC_PASSWORD_MUST_NOT_APPEAR"
+        with self.assertLogs(
+            "config.graphql_auth0_auth.backends", level="DEBUG"
+        ) as logs:
+            Auth0RemoteUserJSONWebTokenBackend().authenticate(
+                None, username="synthetic", password=secret
+            )
+        self.assertNotIn(secret, "\n".join(logs.output))
diff --git a/opencontractserver/tests/test_document_imports_worker_token.py b/opencontractserver/tests/test_document_imports_worker_token.py
index a377732741..d72c9d8348 100644
--- a/opencontractserver/tests/test_document_imports_worker_token.py
+++ b/opencontractserver/tests/test_document_imports_worker_token.py
@@ -289,19 +289,19 @@ def test_complete_accepts_matching_token(self):
 
     def test_complete_rejects_swapped_token(self):
         """A token bound to a *different* corpus cannot complete a session that
-        another token started (token-swap guard); the session is marked FAILED."""
+        another token started (token-swap guard); the session is left untouched."""
         session = self._start_and_fill_zip_session(self.token)
         swapped, _ = CorpusAccessToken.create_token(
             worker_account=self.account, corpus=self.other
         )
-        with self.assertRaises(DocumentImportPermissionError):
+        with self.assertRaises(ChunkedUploadError):
             complete_chunked_upload(
                 user=self.account.user,
                 upload_id=session.id,
                 access_token=swapped,
             )
         session.refresh_from_db()
-        self.assertEqual(session.status, ChunkedUploadStatus.FAILED)
+        self.assertEqual(session.status, ChunkedUploadStatus.PENDING)
 
     def test_complete_rejects_swapped_token_document_kind(self):
         """The completion token-rebind guard also protects DOCUMENT-kind
@@ -325,14 +325,14 @@ def test_complete_rejects_swapped_token_document_kind(self):
         swapped, _ = CorpusAccessToken.create_token(
             worker_account=self.account, corpus=self.other
         )
-        with self.assertRaises(DocumentImportPermissionError):
+        with self.assertRaises(ChunkedUploadError):
             complete_chunked_upload(
                 user=self.account.user,
                 upload_id=session.id,
                 access_token=swapped,
             )
         session.refresh_from_db()
-        self.assertEqual(session.status, ChunkedUploadStatus.FAILED)
+        self.assertEqual(session.status, ChunkedUploadStatus.PENDING)
 
 
 @override_settings(CELERY_TASK_ALWAYS_EAGER=False)
@@ -368,6 +368,55 @@ def _client(self) -> APIClient:
     def _zip_upload(self):
         return io.BytesIO(_zip_bytes())
 
+    def test_chunked_operations_enforce_corpus_scope_for_same_worker(self):
+        from tempfile import TemporaryDirectory
+
+        other_token, other_plaintext = CorpusAccessToken.create_token(
+            worker_account=self.account, corpus=self.other
+        )
+        with TemporaryDirectory() as media_root, override_settings(
+            MEDIA_ROOT=media_root
+        ):
+            session = start_chunked_upload(
+                user=self.account.user,
+                kind="document",
+                filename="x.pdf",
+                total_size=len(_PDF),
+                chunk_size=len(_PDF),
+                total_chunks=1,
+                metadata={"title": "x.pdf"},
+                access_token=other_token,
+            )
+            self.assertEqual(session.metadata["add_to_corpus_id"], str(self.other.pk))
+            store_chunk(
+                user=self.account.user,
+                upload_id=session.pk,
+                index=0,
+                chunk_file=SimpleUploadedFile("x.pdf", _PDF),
+                access_token=other_token,
+            )
+            root = f"/api/imports/chunked/{session.pk}/"
+            wrong = self._client()
+            self.assertEqual(wrong.get(root).status_code, 404)
+            self.assertEqual(
+                wrong.post(
+                    root + "parts/0/",
+                    {"file": SimpleUploadedFile("x.pdf", b"x" * len(_PDF))},
+                    format="multipart",
+                ).status_code,
+                404,
+            )
+            self.assertEqual(
+                wrong.post(root + "complete/", {}, format="json").status_code, 404
+            )
+            session.refresh_from_db()
+            self.assertEqual(session.status, ChunkedUploadStatus.PENDING)
+            with session.parts.get(index=0).file.open("rb") as part:
+                self.assertEqual(part.read(), _PDF)
+            allowed = APIClient()
+            allowed.credentials(HTTP_AUTHORIZATION=f"WorkerKey {other_plaintext}")
+            self.assertEqual(allowed.get(root).status_code, 200)
+
     def test_workerkey_zip_to_corpus_accepted(self):
         r = self._client().post(
             "/api/imports/zip-to-corpus/",
diff --git a/opencontractserver/tests/test_graphql_resource_contracts.py b/opencontractserver/tests/test_graphql_resource_contracts.py
new file mode 100644
index 0000000000..5f5b8a6e29
--- /dev/null
+++ b/opencontractserver/tests/test_graphql_resource_contracts.py
@@ -0,0 +1,468 @@
+"""GraphQL resource visibility and update contracts."""
+
+import json
+from typing import Any
+
+from django.contrib.auth import get_user_model
+from django.contrib.auth.models import AnonymousUser
+from django.test import TestCase
+
+from config.jwt_auth.shortcuts import get_token
+from opencontractserver.agents.models import AgentConfiguration
+from opencontractserver.analyzer.models import Analysis, Analyzer, GremlinEngine
+from opencontractserver.annotations.models import (
+    Annotation,
+    AnnotationLabel,
+    LabelSet,
+    Note,
+    Relationship,
+)
+from opencontractserver.corpuses.models import Corpus, CorpusAction, CorpusActionTrigger
+from opencontractserver.documents.models import Document, DocumentPath
+from opencontractserver.extracts.models import Fieldset
+from opencontractserver.shared.services.tree_traversal import TreeTraversalService
+from opencontractserver.types.enums import PermissionTypes
+from opencontractserver.utils.ids import to_global_id
+from opencontractserver.utils.permissioning import set_permissions_for_obj_to_user
+
+User = get_user_model()
+
+
+class GraphQLResourceContractTests(TestCase):
+    def setUp(self):
+        self.owner = User.objects.create_user(username="audit_owner", password="audit")
+        self.viewer = User.objects.create_user(
+            username="audit_viewer", password="audit"
+        )
+        self.corpus = Corpus.objects.create(
+            title="Audit corpus", creator=self.owner, is_public=True
+        )
+        self.document = Document.objects.create(
+            title="Audit document", creator=self.owner, is_public=True
+        )
+        DocumentPath.objects.create(
+            document=self.document,
+            corpus=self.corpus,
+            creator=self.owner,
+            path="audit.pdf",
+            version_number=1,
+            is_current=True,
+            is_deleted=False,
+        )
+
+    def query(self, query, variables, user=None):
+        headers: dict[str, Any] = (
+            {"HTTP_AUTHORIZATION": f"Bearer {get_token(user)}"} if user else {}
+        )
+        response = self.client.post(
+            "/graphql/",
+            {"query": query, "variables": variables},
+            content_type="application/json",
+            **headers,
+        )
+        self.assertEqual(response.status_code, 200, response.content)
+        result = response.json()
+        self.assertNotIn("errors", result, result)
+        return result["data"]
+
+    def test_corpus_update_preserves_owner(self):
+        set_permissions_for_obj_to_user(
+            self.viewer, self.corpus, [PermissionTypes.READ, PermissionTypes.UPDATE]
+        )
+        self.assertFalse(self.corpus.user_can(self.viewer, PermissionTypes.DELETE))
+        result = self.query(
+            'mutation($id: String!) { updateCorpus(id: $id, title: "Edited") { ok message } }',
+            {"id": to_global_id("CorpusType", self.corpus.pk)},
+            self.viewer,
+        )
+        self.assertTrue(result["updateCorpus"]["ok"], result)
+        self.corpus.refresh_from_db()
+        self.assertEqual(
+            self.corpus.creator_id,
+            self.owner.pk,
+            "UPDATE transferred ownership to the editor",
+        )
+        fresh = Corpus.objects.get(pk=self.corpus.pk)
+        self.assertFalse(fresh.user_can(self.viewer, PermissionTypes.DELETE))
+        self.assertFalse(fresh.user_can(self.viewer, PermissionTypes.PERMISSION))
+
+    def test_labelset_update_preserves_owner(self):
+        labelset = LabelSet.objects.create(title="Audit labels", creator=self.owner)
+        set_permissions_for_obj_to_user(
+            self.viewer, labelset, [PermissionTypes.READ, PermissionTypes.UPDATE]
+        )
+        result = self.query(
+            'mutation($id: String!) { updateLabelset(id: $id, title: "Edited") { ok message } }',
+            {"id": to_global_id("LabelSetType", labelset.pk)},
+            self.viewer,
+        )
+        self.assertTrue(result["updateLabelset"]["ok"], result)
+        labelset.refresh_from_db()
+        self.assertEqual(
+            labelset.creator_id, self.owner.pk, "UPDATE transferred label-set ownership"
+        )
+
+    def test_labelset_assignment_requires_read(self):
+        labelset = LabelSet.objects.create(
+            title="SYNTHETIC_PRIVATE_LABELS", creator=self.owner
+        )
+        set_permissions_for_obj_to_user(
+            self.viewer, self.corpus, [PermissionTypes.READ, PermissionTypes.UPDATE]
+        )
+        query = "mutation($id: String!, $labels: String!) { updateCorpus(id: $id, labelSet: $labels) { ok } }"
+        variables = {
+            "id": to_global_id("CorpusType", self.corpus.pk),
+            "labels": to_global_id("LabelSetType", labelset.pk),
+        }
+        result = self.query(query, variables, self.viewer)
+        self.assertFalse(result["updateCorpus"]["ok"])
+        self.corpus.refresh_from_db()
+        self.assertIsNone(self.corpus.label_set_id)
+        set_permissions_for_obj_to_user(self.viewer, labelset, [PermissionTypes.READ])
+        result = self.query(query, variables, self.viewer)
+        self.assertTrue(result["updateCorpus"]["ok"])
+        self.corpus.refresh_from_db()
+        self.assertEqual(self.corpus.label_set_id, labelset.pk)
+        self.assertEqual(self.corpus.creator_id, self.owner.pk)
+
+    def privacy_fixture(self):
+        analyzer = Analyzer.objects.create(
+            id="audit_analyzer",
+            creator=self.owner,
+            task_name="opencontractserver.tasks.noop",
+        )
+        analysis = Analysis.objects.create(
+            analyzer=analyzer,
+            analyzed_corpus=self.corpus,
+            creator=self.owner,
+            is_public=False,
+        )
+        label = AnnotationLabel.objects.create(
+            text="Audit", creator=self.owner, label_type="TOKEN_LABEL"
+        )
+        common = dict(
+            creator=self.owner,
+            document=self.document,
+            corpus=self.corpus,
+            page=1,
+            json={},
+            annotation_label=label,
+        )
+        plain = Annotation.objects.create(
+            raw_text="public annotation", structural=True, **common
+        )
+        private = Annotation.objects.create(
+            raw_text="AUDIT_SECRET_ANALYSIS_TEXT",
+            created_by_analysis=analysis,
+            parent=plain,
+            **common,
+        )
+        self.assertTrue(
+            Annotation.objects.visible_to_user(AnonymousUser())
+            .filter(pk=plain.pk)
+            .exists()
+        )
+        self.assertFalse(
+            Annotation.objects.visible_to_user(AnonymousUser())
+            .filter(pk=private.pk)
+            .exists()
+        )
+        return plain, private, analysis
+
+    def test_tree_fields_respect_source_visibility(self):
+        plain, private, _ = self.privacy_fixture()
+        result = self.query(
+            "query($id: ID!) { annotation(id: $id) { id descendantsTree fullTree subtree } }",
+            {"id": to_global_id("AnnotationType", plain.pk)},
+        )
+        self.assertNotIn(
+            private.raw_text,
+            json.dumps(result),
+            "Anonymous tree traversal returned analysis-private text",
+        )
+
+    def test_tree_keeps_private_source_content_for_authorized_owner(self):
+        plain, private, _ = self.privacy_fixture()
+        result = self.query(
+            "query($id: ID!) { annotation(id: $id) { descendantsTree fullTree subtree } }",
+            {"id": to_global_id("AnnotationType", plain.pk)},
+            self.owner,
+        )
+        self.assertIn(private.raw_text, json.dumps(result))
+
+    def test_note_trees_and_links_respect_independent_visibility(self):
+        _, private_annotation, _ = self.privacy_fixture()
+        private_document = Document.objects.create(
+            title="Private note document", creator=self.owner
+        )
+        ancestor = Note.objects.create(
+            title="Private ancestor",
+            content="PRIVATE_ANCESTOR",
+            document=private_document,
+            corpus=self.corpus,
+            creator=self.owner,
+        )
+        root = Note.objects.create(
+            title="Visible root",
+            content="VISIBLE_ROOT",
+            parent=ancestor,
+            document=self.document,
+            corpus=self.corpus,
+            creator=self.owner,
+            annotation=private_annotation,
+        )
+        Note.objects.create(
+            title="Private child",
+            content="PRIVATE_CHILD",
+            parent=root,
+            document=private_document,
+            corpus=self.corpus,
+            creator=self.owner,
+        )
+        Note.objects.create(
+            title="Visible child",
+            content="VISIBLE_CHILD",
+            parent=root,
+            document=self.document,
+            corpus=self.corpus,
+            creator=self.owner,
+        )
+        query = """query($id: ID!) { note(id: $id) {
+            parent { id } corpus { id } annotation { id }
+            descendantsTree fullTree subtree
+        } }"""
+        variables = {"id": to_global_id("NoteType", root.pk)}
+        result = self.query(query, variables, self.viewer)["note"]
+        self.assertIsNone(result["parent"])
+        self.assertIsNone(result["annotation"])
+        self.assertEqual(
+            result["corpus"]["id"], to_global_id("CorpusType", self.corpus.pk)
+        )
+        for field in ("descendantsTree", "fullTree", "subtree"):
+            with self.subTest(field=field):
+                tree = json.dumps(result[field])
+                self.assertIn("VISIBLE_CHILD", tree)
+                self.assertNotIn("PRIVATE_CHILD", tree)
+                self.assertNotIn("PRIVATE_ANCESTOR", tree)
+
+        result = self.query(query, variables, self.owner)["note"]
+        self.assertEqual(result["parent"]["id"], to_global_id("NoteType", ancestor.pk))
+        self.assertEqual(
+            result["annotation"]["id"],
+            to_global_id("AnnotationType", private_annotation.pk),
+        )
+        self.assertIn("PRIVATE_ANCESTOR", json.dumps(result["fullTree"]))
+        self.assertIn("PRIVATE_CHILD", json.dumps(result["descendantsTree"]))
+
+    def test_tree_traversal_rejects_hidden_roots_and_terminates_cycles(self):
+        plain, private, _ = self.privacy_fixture()
+        self.assertEqual(
+            TreeTraversalService.get_nodes(
+                private, self.viewer, mode="full", text_field="raw_text"
+            ),
+            [],
+        )
+        Annotation.objects.filter(pk=plain.pk).update(parent=private)
+        plain.refresh_from_db()
+        for mode in ("full", "subtree", "descendants"):
+            with self.subTest(mode=mode):
+                nodes = TreeTraversalService.get_nodes(
+                    plain, self.owner, mode=mode, text_field="raw_text"
+                )
+                expected = (
+                    {private.pk} if mode == "descendants" else {plain.pk, private.pk}
+                )
+                self.assertEqual({node["id"] for node in nodes}, expected)
+                self.assertEqual(len(nodes), len(expected))
+
+    def test_action_configuration_links_require_target_visibility(self):
+        private_corpus = Corpus.objects.create(
+            title="Private config corpus", creator=self.owner
+        )
+        resources = {
+            "fieldset": Fieldset.objects.create(
+                name="Private fieldset", creator=self.owner
+            ),
+            "analyzer": Analyzer.objects.create(
+                id="private_action_analyzer",
+                creator=self.owner,
+                task_name="opencontractserver.tasks.noop",
+            ),
+            "agent_config": AgentConfiguration.objects.create(
+                name="Private config",
+                scope="CORPUS",
+                corpus=private_corpus,
+                creator=self.owner,
+                system_instructions="Synthetic instructions",
+            ),
+        }
+        actions = {}
+        for field, resource in resources.items():
+            action = CorpusAction.objects.create(
+                name=field,
+                corpus=self.corpus,
+                creator=self.owner,
+                trigger=CorpusActionTrigger.ADD_DOCUMENT,
+                task_instructions="Synthetic task" if field == "agent_config" else "",
+                **{field: resource},
+            )
+            set_permissions_for_obj_to_user(self.viewer, action, [PermissionTypes.READ])
+            actions[to_global_id("CorpusActionType", action.pk)] = field
+
+        query = """query($corpusId: ID) { corpusActions(corpusId: $corpusId) {
+            edges { node { id fieldset { id } analyzer { id } agentConfig { id } } }
+        } }"""
+        variables = {"corpusId": to_global_id("CorpusType", self.corpus.pk)}
+        viewer_nodes = self.query(query, variables, self.viewer)["corpusActions"][
+            "edges"
+        ]
+        self.assertEqual(len(viewer_nodes), 3)
+        for edge in viewer_nodes:
+            self.assertIsNone(edge["node"]["fieldset"])
+            self.assertIsNone(edge["node"]["analyzer"])
+            self.assertIsNone(edge["node"]["agentConfig"])
+
+        owner_nodes = self.query(query, variables, self.owner)["corpusActions"]["edges"]
+        self.assertEqual(len(owner_nodes), 3)
+        types = {
+            "fieldset": "FieldsetType",
+            "analyzer": "AnalyzerType",
+            "agent_config": "AgentConfigurationType",
+        }
+        for edge in owner_nodes:
+            node = edge["node"]
+            field = actions[node["id"]]
+            api_field = "agentConfig" if field == "agent_config" else field
+            self.assertEqual(
+                node[api_field]["id"], to_global_id(types[field], resources[field].pk)
+            )
+
+    def test_legacy_analysis_link_obeys_independent_visibility(self):
+        plain, _, analysis = self.privacy_fixture()
+        plain.analysis = analysis
+        plain.save(update_fields=["analysis"])
+        query = "query($id: ID!) { annotation(id: $id) { analysis { id } } }"
+        variables = {"id": to_global_id("AnnotationType", plain.pk)}
+        self.assertIsNone(self.query(query, variables)["annotation"]["analysis"])
+        self.assertEqual(
+            self.query(query, variables, self.owner)["annotation"]["analysis"]["id"],
+            to_global_id("AnalysisType", analysis.pk),
+        )
+
+    def test_relationship_lists_respect_visibility(self):
+        plain, _, analysis = self.privacy_fixture()
+        label = AnnotationLabel.objects.create(
+            text="AUDIT_SECRET_RELATIONSHIP",
+            creator=self.owner,
+            label_type="RELATIONSHIP_LABEL",
+        )
+        relation = Relationship.objects.create(
+            creator=self.owner,
+            document=self.document,
+            corpus=self.corpus,
+            created_by_analysis=analysis,
+            analysis=analysis,
+            analyzer=analysis.analyzer,
+            relationship_label=label,
+        )
+        relation.source_annotations.add(plain)
+        relation.target_annotations.add(plain)
+        self.assertFalse(
+            Relationship.objects.visible_to_user(AnonymousUser())
+            .filter(pk=relation.pk)
+            .exists()
+        )
+        query = """query($id: ID!) { annotation(id: $id) {
+            allSourceNodeInRelationship {
+                id relationshipLabel { text } analyzer { id }
+                analysis { id } createdByAnalysis { id } createdByExtract { id }
+            }
+            allTargetNodeInRelationship { id }
+        } }"""
+        variables = {"id": to_global_id("AnnotationType", plain.pk)}
+        result = self.query(query, variables)
+        self.assertNotIn(
+            label.text,
+            json.dumps(result),
+            "Anonymous relationship traversal bypassed source privacy",
+        )
+        self.assertEqual(result["annotation"]["allTargetNodeInRelationship"], [])
+        result = self.query(query, variables, self.owner)["annotation"]
+        expected_id = to_global_id("RelationshipType", relation.pk)
+        self.assertEqual(result["allTargetNodeInRelationship"], [{"id": expected_id}])
+        visible = result["allSourceNodeInRelationship"][0]
+        self.assertEqual(visible["id"], expected_id)
+        self.assertEqual(
+            visible["analysis"]["id"], to_global_id("AnalysisType", analysis.pk)
+        )
+        self.assertEqual(visible["createdByAnalysis"], visible["analysis"])
+        self.assertEqual(
+            visible["analyzer"]["id"],
+            to_global_id("AnalyzerType", analysis.analyzer_id),
+        )
+        self.assertIsNone(visible["createdByExtract"])
+
+    def test_parent_annotation_respects_visibility(self):
+        plain, private, _ = self.privacy_fixture()
+        private.parent = None
+        private.save(update_fields=["parent"])
+        plain.parent = private
+        plain.save(update_fields=["parent"])
+        result = self.query(
+            "query($id: ID!) { annotation(id: $id) { parent { id rawText } } }",
+            {"id": to_global_id("AnnotationType", plain.pk)},
+        )
+        self.assertIsNone(
+            result["annotation"]["parent"],
+            "Default FK resolution exposed a private parent",
+        )
+
+    def test_analyzer_engine_respects_visibility(self):
+        engine = GremlinEngine.objects.create(
+            creator=self.owner,
+            url="https://example.invalid",
+            api_key="SYNTHETIC_ENGINE_SECRET",
+        )
+        analyzer = Analyzer.objects.create(
+            id="remote_audit", creator=self.owner, host_gremlin=engine, is_public=True
+        )
+        result = self.query(
+            "query($id: ID!) { analyzer(id: $id) { hostGremlin { url apiKey } } }",
+            {"id": to_global_id("AnalyzerType", analyzer.pk)},
+        )
+        self.assertIsNone(result["analyzer"]["hostGremlin"])
+
+    def test_engine_credentials_require_management_access(self):
+        engine = GremlinEngine.objects.create(
+            creator=self.owner,
+            is_public=True,
+            url="https://example.invalid",
+            api_key="SYNTHETIC_ENGINE_SECRET",
+        )
+        analyzer = Analyzer.objects.create(
+            id="remote_audit", creator=self.owner, host_gremlin=engine, is_public=True
+        )
+        query = "query($id: ID!) { analyzer(id: $id) { hostGremlin { url apiKey } } }"
+        variables = {"id": to_global_id("AnalyzerType", analyzer.pk)}
+        result = self.query(query, variables)
+        self.assertEqual(result["analyzer"]["hostGremlin"]["url"], engine.url)
+        self.assertIsNone(result["analyzer"]["hostGremlin"]["apiKey"])
+        result = self.query(query, variables, self.owner)
+        self.assertEqual(result["analyzer"]["hostGremlin"]["apiKey"], engine.api_key)
+
+    def test_corpus_labelset_respects_visibility(self):
+        labelset = LabelSet.objects.create(
+            title="SYNTHETIC_PRIVATE_LABELSET", creator=self.owner
+        )
+        self.corpus.label_set = labelset
+        self.corpus.save(update_fields=["label_set"])
+        self.assertFalse(
+            LabelSet.objects.visible_to_user(AnonymousUser())
+            .filter(pk=labelset.pk)
+            .exists()
+        )
+        result = self.query(
+            "query($id: ID!) { corpus(id: $id) { labelSet { id title } } }",
+            {"id": to_global_id("CorpusType", self.corpus.pk)},
+        )
+        self.assertIsNone(result["corpus"]["labelSet"])
diff --git a/opencontractserver/tests/test_pydantic_ai_tools_module.py b/opencontractserver/tests/test_pydantic_ai_tools_module.py
index aa0dd1cefc..53329ec99f 100644
--- a/opencontractserver/tests/test_pydantic_ai_tools_module.py
+++ b/opencontractserver/tests/test_pydantic_ai_tools_module.py
@@ -266,6 +266,14 @@ def setUpTestData(cls):
         )
         cls.doc, _, _ = cls.corpus.add_document(document=original_doc, user=cls.user)
 
+    async def test_disabled_user_loses_tool_access_mid_session(self):
+        deps = PydanticAIDependencies(user_id=self.user.pk, corpus_id=self.corpus.pk)
+        ctx = MagicMock(deps=deps)
+        await _check_user_permissions(ctx)
+        await User.objects.filter(pk=self.user.pk).aupdate(is_active=False)
+        with self.assertRaisesRegex(PermissionError, "inactive"):
+            await _check_user_permissions(ctx)
+
     async def test_anonymous_user_nonexistent_document_raises_error(self):
         """Test that anonymous user accessing non-existent document raises PermissionError.
 
diff --git a/opencontractserver/tests/test_security_hardening.py b/opencontractserver/tests/test_security_hardening.py
index 736bd17350..373f7cadbd 100644
--- a/opencontractserver/tests/test_security_hardening.py
+++ b/opencontractserver/tests/test_security_hardening.py
@@ -1199,6 +1199,40 @@ class TestDepthLimitValidationRule(TestCase):
     does not pass validation_rules (those are applied by GraphQLView in urls.py).
     """
 
+    def test_fragment_dag_is_measured_without_exponential_expansion(self):
+        from types import SimpleNamespace
+        from unittest.mock import Mock
+
+        from graphql import parse
+
+        from config.graphql.security import _measure_depth
+
+        definitions = ["query { ...F0 }"]
+        for index in range(20):
+            nested = f"...F{index + 1} ...F{index + 1}" if index < 19 else "rawText"
+            definitions.append(
+                f"fragment F{index} on AnnotationType {{ parent {{ {nested} }} }}"
+            )
+        document = parse("\n".join(definitions))
+        fragments = {node.name.value: node for node in document.definitions[1:]}
+        context = SimpleNamespace(get_fragment=Mock(side_effect=fragments.get))
+        self.assertEqual(_measure_depth(document.definitions[0], context=context), 21)
+        self.assertEqual(context.get_fragment.call_count, 20)
+
+    def test_reused_fragment_is_counted_at_each_depth(self):
+        query = """
+            query {
+                annotation(id: "unused") {
+                    ...Leaf
+                    parent { parent { ...Leaf } }
+                }
+            }
+            fragment Leaf on AnnotationType { parent { rawText } }
+        """
+        errors = self._validate_query(query, max_depth=4)
+        self.assertTrue(any("depth" in str(error).lower() for error in errors))
+        self.assertFalse(self._validate_query(query, max_depth=5))
+
     def _validate_query(self, query_str, max_depth):
         """Validate a query against the real schema with a given depth limit."""
         from graphql import parse, validate
diff --git a/opencontractserver/tests/test_websocket_session_contracts.py b/opencontractserver/tests/test_websocket_session_contracts.py
new file mode 100644
index 0000000000..1b21de107c
--- /dev/null
+++ b/opencontractserver/tests/test_websocket_session_contracts.py
@@ -0,0 +1,392 @@
+"""WebSocket admission and session lifecycle contracts."""
+
+import time
+from datetime import datetime, timedelta, timezone
+from unittest.mock import patch
+
+from asgiref.sync import async_to_sync
+from channels.db import database_sync_to_async
+from channels.layers import get_channel_layer
+from channels.testing import WebsocketCommunicator
+from django.contrib.auth import get_user_model
+from django.test import TransactionTestCase
+
+from config.jwt_auth.shortcuts import get_token
+from config.websocket.auth_handshake import _get_user_from_token
+from config.websocket.consumers.thread_updates import (
+    ThreadUpdatesConsumer,
+    get_thread_channel_group,
+)
+from config.websocket.middleware import WS_AUTH_SUBPROTOCOL, JWTAuthMiddleware
+from opencontractserver.conversations.models import (
+    Conversation,
+    ConversationTypeChoices,
+)
+from opencontractserver.conversations.services import ConversationService
+from opencontractserver.corpuses.models import Corpus
+
+User = get_user_model()
+
+
+class WebSocketSessionContractTests(TransactionTestCase):
+    def setUp(self):
+        self.owner = User.objects.create_user(
+            username="audit_ws_owner", password="audit"
+        )
+        self.viewer = User.objects.create_user(
+            username="audit_ws_viewer", password="audit"
+        )
+        self.corpus = Corpus.objects.create(
+            title="Audit ws corpus", creator=self.owner, is_public=True
+        )
+
+    def conversation(self, kind):
+        return Conversation.objects.create(
+            creator=self.owner,
+            chat_with_corpus=self.corpus,
+            conversation_type=kind,
+            is_public=False,
+        )
+
+    def test_socket_uses_conversation_visibility(self):
+        conversation = self.conversation(ConversationTypeChoices.CHAT)
+        self.assertFalse(
+            Conversation.objects.visible_to_user(self.viewer)
+            .filter(pk=conversation.pk)
+            .exists()
+        )
+        consumer = ThreadUpdatesConsumer()
+        consumer.conversation_id = conversation.pk
+        allowed = async_to_sync(consumer._check_conversation_access)(
+            self.viewer, cache_conversation=False
+        )
+        self.assertFalse(
+            allowed, "Corpus READ improperly grants access to a private CHAT socket"
+        )
+
+    def test_administrator_uses_conversation_visibility(self):
+        self.corpus.is_public = False
+        self.corpus.save(update_fields=["is_public"])
+        conversation = self.conversation(ConversationTypeChoices.THREAD)
+        self.viewer.is_superuser = True
+        self.viewer.save(update_fields=["is_superuser"])
+        self.assertFalse(
+            Conversation.objects.visible_to_user(self.viewer)
+            .filter(pk=conversation.pk)
+            .exists()
+        )
+        consumer = ThreadUpdatesConsumer()
+        consumer.conversation_id = conversation.pk
+        allowed = async_to_sync(consumer._check_conversation_access)(
+            self.viewer, cache_conversation=False
+        )
+        self.assertFalse(
+            allowed,
+            "Socket retained blanket admin access removed from canonical policy",
+        )
+
+    async def connect_socket(self, conversation, token):
+        communicator = WebsocketCommunicator(
+            JWTAuthMiddleware(ThreadUpdatesConsumer.as_asgi()),
+            f"/ws/thread-updates/?conversation_id={conversation.pk}",
+            subprotocols=[WS_AUTH_SUBPROTOCOL, token],
+        )
+        connected, _ = await communicator.connect()
+        self.assertTrue(connected)
+        await communicator.receive_json_from()
+        await communicator.receive_json_from()
+        return communicator
+
+    def test_stream_bursts_bound_permission_queries_per_viewer(self):
+        conversation = self.conversation(ConversationTypeChoices.THREAD)
+        token = get_token(self.viewer)
+
+        async def scenario():
+            viewers = [await self.connect_socket(conversation, token) for _ in range(2)]
+
+            async def broadcast_token():
+                await get_channel_layer().group_send(
+                    get_thread_channel_group(conversation.pk),
+                    {"type": "agent_stream_token", "token": "synthetic token"},
+                )
+
+            async def receive_tokens():
+                for viewer in viewers:
+                    event = await viewer.receive_json_from()
+                    self.assertEqual(event["type"], "AGENT_STREAM_TOKEN")
+
+            try:
+                with (
+                    patch("config.websocket.auth_handshake.time", wraps=time) as clock,
+                    patch(
+                        "config.websocket.auth_handshake._get_user_from_token",
+                        wraps=_get_user_from_token,
+                    ) as load_user,
+                    patch.object(
+                        ConversationService,
+                        "get_or_none",
+                        wraps=ConversationService.get_or_none,
+                    ) as load_conversation,
+                ):
+                    # Exercise real JWT and database visibility checks, with two
+                    # subscribers receiving the same 25-token channel burst.
+                    for index in range(25):
+                        clock.monotonic.return_value = 100 + index / 100
+                        await broadcast_token()
+                        await receive_tokens()
+                    self.assertEqual(load_user.await_count, 2)
+                    self.assertEqual(load_conversation.call_count, 2)
+
+                    # Token activity must not extend the one-second window.
+                    clock.monotonic.return_value = 101.01
+                    await broadcast_token()
+                    await receive_tokens()
+                    self.assertEqual(load_user.await_count, 4)
+                    self.assertEqual(load_conversation.call_count, 4)
+
+                    await database_sync_to_async(
+                        Corpus.objects.filter(pk=self.corpus.pk).update
+                    )(is_public=False)
+                    clock.monotonic.return_value = 102.02
+                    await broadcast_token()
+                    for viewer in viewers:
+                        self.assertEqual(
+                            await viewer.receive_json_from(),
+                            {"type": "AUTH_FAILED", "reason": "PERMISSION_REVOKED"},
+                        )
+                        self.assertEqual((await viewer.receive_output())["code"], 4003)
+                    self.assertEqual(load_user.await_count, 6)
+                    self.assertEqual(load_conversation.call_count, 6)
+            finally:
+                for viewer in viewers:
+                    await viewer.disconnect()
+
+        async_to_sync(scenario)()
+
+    def test_stream_window_never_delays_token_expiration(self):
+        conversation = self.conversation(ConversationTypeChoices.THREAD)
+        expiry = int(time.time()) + 30
+        token = get_token(self.viewer, exp=expiry)
+
+        async def scenario():
+            communicator = await self.connect_socket(conversation, token)
+            try:
+                with patch("config.websocket.auth_handshake.time", wraps=time) as clock:
+                    clock.monotonic.return_value = 100
+                    event = {"type": "agent_stream_token", "token": "synthetic token"}
+                    group = get_thread_channel_group(conversation.pk)
+                    await get_channel_layer().group_send(group, event)
+                    self.assertEqual(
+                        (await communicator.receive_json_from())["type"],
+                        "AGENT_STREAM_TOKEN",
+                    )
+                    clock.monotonic.return_value = 100.1
+                    clock.time.return_value = expiry
+                    await get_channel_layer().group_send(group, event)
+                    self.assertEqual(
+                        await communicator.receive_json_from(),
+                        {"type": "AUTH_FAILED", "reason": "EXPIRED"},
+                    )
+                    self.assertEqual(
+                        (await communicator.receive_output())["code"], 4001
+                    )
+            finally:
+                await communicator.disconnect()
+
+        async_to_sync(scenario)()
+
+    def _assert_stream_window_does_not_delay_revocation(self, *, client_frame):
+        conversation = self.conversation(ConversationTypeChoices.THREAD)
+        token = get_token(self.viewer)
+
+        async def scenario():
+            communicator = await self.connect_socket(conversation, token)
+            try:
+                with patch("config.websocket.auth_handshake.time", wraps=time) as clock:
+                    clock.monotonic.return_value = 100
+                    group = get_thread_channel_group(conversation.pk)
+                    await get_channel_layer().group_send(
+                        group,
+                        {"type": "agent_stream_token", "token": "synthetic token"},
+                    )
+                    self.assertEqual(
+                        (await communicator.receive_json_from())["type"],
+                        "AGENT_STREAM_TOKEN",
+                    )
+                    await database_sync_to_async(
+                        Corpus.objects.filter(pk=self.corpus.pk).update
+                    )(is_public=False)
+                    clock.monotonic.return_value = 100.1
+                    if client_frame:
+                        # A client-controlled type must not opt into token caching.
+                        await communicator.send_json_to({"type": "agent_stream_token"})
+                    else:
+                        await get_channel_layer().group_send(
+                            group,
+                            {"type": "agent_stream_complete", "content": "result"},
+                        )
+                    self.assertEqual(
+                        await communicator.receive_json_from(),
+                        {"type": "AUTH_FAILED", "reason": "PERMISSION_REVOKED"},
+                    )
+                    self.assertEqual(
+                        (await communicator.receive_output())["code"], 4003
+                    )
+            finally:
+                await communicator.disconnect()
+
+        async_to_sync(scenario)()
+
+    def test_client_frames_recheck_within_stream_window(self):
+        self._assert_stream_window_does_not_delay_revocation(client_frame=True)
+
+    def test_completion_rechecks_within_stream_window(self):
+        self._assert_stream_window_does_not_delay_revocation(client_frame=False)
+
+    def test_broadcast_rechecks_resource_access(self):
+        conversation = self.conversation(ConversationTypeChoices.THREAD)
+        token = get_token(self.viewer)
+
+        async def scenario():
+            from config.websocket.consumers.thread_updates import (
+                get_thread_channel_group,
+            )
+
+            communicator = await self.connect_socket(conversation, token)
+            try:
+                await database_sync_to_async(
+                    Corpus.objects.filter(pk=self.corpus.pk).update
+                )(is_public=False)
+                visible = await database_sync_to_async(
+                    Conversation.objects.visible_to_user(self.viewer)
+                    .filter(pk=conversation.pk)
+                    .exists
+                )()
+                self.assertFalse(visible)
+                await get_channel_layer().group_send(
+                    get_thread_channel_group(conversation.pk),
+                    {
+                        "type": "agent_stream_complete",
+                        "content": "AUDIT_SECRET_AFTER_REVOCATION",
+                    },
+                )
+                event = await communicator.receive_output(timeout=2)
+                self.assertNotIn(
+                    "AUDIT_SECRET_AFTER_REVOCATION",
+                    str(event),
+                    "Socket streamed newly private content without an AUTH frame",
+                )
+            finally:
+                await communicator.disconnect()
+
+        async_to_sync(scenario)()
+
+    def test_broadcast_rechecks_session_expiration(self):
+        conversation = self.conversation(ConversationTypeChoices.THREAD)
+        token = get_token(
+            self.viewer, exp=int(datetime.now(timezone.utc).timestamp()) + 30
+        )
+
+        async def scenario():
+            from config.websocket.consumers.thread_updates import (
+                get_thread_channel_group,
+            )
+
+            communicator = await self.connect_socket(conversation, token)
+            try:
+                future = datetime.now(timezone.utc) + timedelta(minutes=2)
+                with patch("jwt.api_jwt.datetime") as clock:
+                    clock.now.return_value = future
+                    await get_channel_layer().group_send(
+                        get_thread_channel_group(conversation.pk),
+                        {
+                            "type": "agent_stream_complete",
+                            "content": "AUDIT_SECRET_AFTER_EXPIRY",
+                        },
+                    )
+                    event = await communicator.receive_output(timeout=2)
+                self.assertNotIn(
+                    "AUDIT_SECRET_AFTER_EXPIRY",
+                    str(event),
+                    "Socket streamed content after its JWT expired",
+                )
+            finally:
+                await communicator.disconnect()
+
+        async_to_sync(scenario)()
+
+    def test_idle_socket_expires_without_client_frames(self):
+        conversation = self.conversation(ConversationTypeChoices.THREAD)
+        token = get_token(
+            self.viewer, exp=int(datetime.now(timezone.utc).timestamp()) + 3
+        )
+
+        async def scenario():
+            communicator = await self.connect_socket(conversation, token)
+            try:
+                failure = await communicator.receive_json_from(timeout=4)
+                self.assertEqual(failure, {"type": "AUTH_FAILED", "reason": "EXPIRED"})
+                close = await communicator.receive_output(timeout=1)
+                self.assertEqual(close["type"], "websocket.close")
+                self.assertEqual(close["code"], 4001)
+            finally:
+                await communicator.disconnect()
+
+        async_to_sync(scenario)()
+
+    def test_idle_socket_detects_disabled_user(self):
+        conversation = self.conversation(ConversationTypeChoices.THREAD)
+        token = get_token(self.viewer)
+
+        async def scenario():
+            with patch(
+                "config.websocket.auth_handshake._AUTH_RECHECK_INTERVAL_SEC", 0.05
+            ):
+                communicator = await self.connect_socket(conversation, token)
+                try:
+                    await database_sync_to_async(
+                        User.objects.filter(pk=self.viewer.pk).update
+                    )(is_active=False)
+                    failure = await communicator.receive_json_from(timeout=2)
+                    self.assertEqual(failure["type"], "AUTH_FAILED")
+                    close = await communicator.receive_output(timeout=1)
+                    self.assertEqual(close["type"], "websocket.close")
+                finally:
+                    await communicator.disconnect()
+
+        async_to_sync(scenario)()
+
+    def test_successful_refresh_replaces_expiration_deadline(self):
+        from config.websocket.consumers.thread_updates import get_thread_channel_group
+
+        conversation = self.conversation(ConversationTypeChoices.THREAD)
+        now = int(datetime.now(timezone.utc).timestamp())
+        token = get_token(self.viewer, exp=now + 30)
+        refreshed_token = get_token(self.viewer, exp=now + 300)
+
+        async def scenario():
+            communicator = await self.connect_socket(conversation, token)
+            try:
+                await communicator.send_json_to(
+                    {"type": "AUTH", "token": refreshed_token}
+                )
+                ack = await communicator.receive_json_from()
+                self.assertEqual(ack["type"], "AUTH_OK")
+                self.assertTrue(ack["refreshed"])
+                with patch("jwt.api_jwt.datetime") as clock:
+                    clock.now.return_value = datetime.now(timezone.utc) + timedelta(
+                        minutes=2
+                    )
+                    await get_channel_layer().group_send(
+                        get_thread_channel_group(conversation.pk),
+                        {
+                            "type": "agent_stream_complete",
+                            "content": "STILL_AUTHORIZED",
+                        },
+                    )
+                    result = await communicator.receive_json_from(timeout=2)
+                self.assertEqual(result["content"], "STILL_AUTHORIZED")
+            finally:
+                await communicator.disconnect()
+
+        async_to_sync(scenario)()
diff --git a/opencontractserver/tests/test_worker_uploads.py b/opencontractserver/tests/test_worker_uploads.py
index b17de14954..1949be34cd 100644
--- a/opencontractserver/tests/test_worker_uploads.py
+++ b/opencontractserver/tests/test_worker_uploads.py
@@ -281,6 +281,14 @@ def test_valid_auth(self):
         response = client.get("/api/worker-uploads/documents/list/")
         self.assertEqual(response.status_code, 200)
 
+    def test_disabled_linked_user_rejects_worker_token(self):
+        self.account.user.is_active = False
+        self.account.user.save(update_fields=["is_active"])
+        client = APIClient()
+        client.credentials(HTTP_AUTHORIZATION=f"WorkerKey {self.plaintext_key}")
+        response = client.get("/api/worker-uploads/documents/list/")
+        self.assertEqual(response.status_code, 401)
+
     def test_missing_token(self):
         client = APIClient()
         response = client.get("/api/worker-uploads/documents/list/")
diff --git a/opencontractserver/worker_uploads/auth.py b/opencontractserver/worker_uploads/auth.py
index ae4e81df78..98ec68648b 100644
--- a/opencontractserver/worker_uploads/auth.py
+++ b/opencontractserver/worker_uploads/auth.py
@@ -71,7 +71,10 @@ def _authenticate_token(self, plaintext_key: str) -> tuple[Any, CorpusAccessToke
             )
             raise exceptions.AuthenticationFailed("Token has been revoked.")
 
-        if not token.worker_account.is_active:
+        if (
+            not token.worker_account.is_active
+            or not token.worker_account.user.is_active
+        ):
             logger.warning(
                 "WorkerToken auth failed: inactive account %s (prefix=%s)",
                 token.worker_account.name,