diff --git a/.gitignore b/.gitignore
index 638075ac..49dad48e 100644
--- a/.gitignore
+++ b/.gitignore
@@ -70,6 +70,7 @@ shell.nix
 **/.jekyll-cache
 
 .beaker.conf
+.beaker/session-attachments/
 beaker_config.py
 beaker_*_config.py
 src/beaker_notebook/**/ui/*
diff --git a/beaker-ts/src/notebook.ts b/beaker-ts/src/notebook.ts
index f7cc5934..b499c399 100644
--- a/beaker-ts/src/notebook.ts
+++ b/beaker-ts/src/notebook.ts
@@ -196,6 +196,22 @@ export type BeakerQueryEvent =
 export interface IQueryCell extends nbformat.IBaseCell {
     cell_type: 'query';
     events: BeakerQueryEvent[];
+    session_attachments?: ISessionAttachment[];
+}
+
+export interface ISessionAttachment extends PartialJSONObject {
+    id: string;
+    name: string;
+    mimetype: string;
+    size: number;
+    kind: "file" | "archive";
+    path?: string | null;
+    root_path: string;
+    original_path?: string | null;
+    archive_status?: "extracted" | "failed" | null;
+    archive_error?: string | null;
+    file_count: number;
+    files: string[];
 }
 
 export class BeakerRawCell extends BeakerBaseCell implements nbformat.IRawCell {
@@ -335,6 +351,8 @@ export class BeakerMarkdownCell extends BeakerBaseCell implements nbformat.IMark
 export class BeakerQueryCell extends BeakerBaseCell implements IQueryCell {
     cell_type: "query" = "query";
     events: BeakerQueryEvent[] = [];
+    /** Live-session chat attachments. Intentionally omitted from notebook serialization. */
+    session_attachments: ISessionAttachment[] = [];
     _current_input_request_message?: messages.IInputRequestMsg;
     _toolCallIndex: Map<string, {eventIndex: number; slot: number}> = new Map();
 
@@ -564,7 +582,8 @@ export class BeakerQueryCell extends BeakerBaseCell implements IQueryCell {
         const future = session.sendBeakerMessage(
             "llm_request",
             {
-                request: this.source
+                request: this.source,
+                attachments: this.session_attachments.map((attachment) => attachment.id),
             },
             undefined,
             metadata,
diff --git a/beaker-ts/src/session.ts b/beaker-ts/src/session.ts
index a4eaf629..d1c9b4f6 100644
--- a/beaker-ts/src/session.ts
+++ b/beaker-ts/src/session.ts
@@ -10,7 +10,7 @@ import * as nbformat from '@jupyterlab/nbformat';
 import type { ConnectionStatus as JupyterConnectionStatus, IAnyMessageArgs } from '@jupyterlab/services/lib/kernel/kernel';
 
 import { createMessageId, IBeakerAvailableContexts, IBeakerFuture, IActiveContextInfo } from './util';
-import { BeakerNotebook, IBeakerShellMessage, IBeakerAnyMessage, BeakerRawCell, BeakerCodeCell, BeakerMarkdownCell, BeakerQueryCell, IBeakerIOPubMessage } from './notebook';
+import { BeakerNotebook, IBeakerShellMessage, IBeakerAnyMessage, BeakerRawCell, BeakerCodeCell, BeakerMarkdownCell, BeakerQueryCell, IBeakerIOPubMessage, ISessionAttachment } from './notebook';
 import { BeakerHistory } from './history';
 import { BeakerRenderer, type IBeakerRendererOptions } from './render';
 import { ChatHistory, type IChatHistory } from './chatHistory';
@@ -390,12 +390,13 @@ export class BeakerSession {
      * @param metadata - (Optional) Any metadata to be associated with the cell.
      * @returns - A reference to the generated cell
      */
-    public addQueryCell(source: string, metadata={}) {
+    public addQueryCell(source: string, metadata={}, sessionAttachments: ISessionAttachment[] = []) {
         const cell = new BeakerQueryCell({
             cell_type: "query",
             source,
             metadata,
         });
+        cell.session_attachments = sessionAttachments;
         this.notebook.addCell(cell);
         return cell;
     };
diff --git a/beaker-vue/src/components/agent/BeakerAgentQuery.vue b/beaker-vue/src/components/agent/BeakerAgentQuery.vue
index 011d5b07..1f8819dd 100644
--- a/beaker-vue/src/components/agent/BeakerAgentQuery.vue
+++ b/beaker-vue/src/components/agent/BeakerAgentQuery.vue
@@ -1,5 +1,16 @@
 <template>
-    <div id="agent-input">
+    <div
+        id="agent-input"
+        :class="{'attachment-drop-active': isDraggingFiles}"
+        @dragenter="handleDragEnter"
+        @dragleave="handleDragLeave"
+        @dragover="handleDragOver"
+        @drop="handleDrop"
+    >
+        <div v-if="isDraggingFiles" class="attachment-drop-overlay">
+            <i class="pi pi-cloud-upload"></i>
+            <span>Drop files to attach them to this message</span>
+        </div>
         <div id="agent-prompt">
             {{ isAwaitingInput ? 'The agent has a question:' : 'How can the agent help?' }}
         </div>
@@ -8,8 +19,74 @@
             <div class="question-text" v-html="questionMarkdown"></div>
         </div>
 
+        <div v-if="draftAttachments.length" class="attachment-drafts" aria-label="Message attachments">
+            <div
+                v-for="draft in draftAttachments"
+                :key="draft.localId"
+                class="attachment-draft"
+                data-testid="attachment-draft"
+                :class="`attachment-${draft.status}`"
+                :title="draft.error || draft.attachment?.archive_error || draft.file.name"
+            >
+                <i
+                    class="attachment-status-icon"
+                    :class="draft.status === 'uploading'
+                        ? 'pi pi-spin pi-spinner'
+                        : draft.status === 'error'
+                            ? 'pi pi-exclamation-triangle'
+                            : draft.attachment?.kind === 'archive'
+                                ? 'pi pi-box'
+                                : 'pi pi-paperclip'"
+                ></i>
+                <span class="attachment-name">{{ draft.file.name }}</span>
+                <small>{{ formatAttachmentSize(draft.file.size) }}</small>
+                <small
+                    v-if="draft.attachment?.kind === 'archive' && draft.attachment.archive_status === 'extracted'"
+                    class="attachment-detail"
+                >
+                    {{ draft.attachment.file_count }} {{ draft.attachment.file_count === 1 ? 'file' : 'files' }}
+                </small>
+                <small v-if="draft.status === 'error'" class="attachment-error">Upload failed</small>
+                <small
+                    v-if="draft.attachment?.archive_status === 'failed'"
+                    class="attachment-error"
+                >
+                    ZIP kept; extraction failed
+                </small>
+                <Button
+                    type="button"
+                    icon="pi pi-times"
+                    text
+                    rounded
+                    size="small"
+                    class="attachment-remove"
+                    :aria-label="`Remove ${draft.file.name}`"
+                    @click="removeAttachment(draft)"
+                />
+            </div>
+        </div>
+
         <div id="agent-inner-input">
             <div class="query-input-container">
+                <input
+                    ref="fileInputRef"
+                    type="file"
+                    multiple
+                    class="attachment-file-input"
+                    data-testid="attachment-file-input"
+                    @change="handleFileSelection"
+                />
+                <Button
+                    v-if="!isAwaitingInput"
+                    type="button"
+                    icon="pi pi-paperclip"
+                    text
+                    rounded
+                    class="attachment-picker"
+                    aria-label="Attach files"
+                    v-tooltip.top="'Attach temporary files'"
+                    @click="fileInputRef?.click()"
+                />
                 <ContainedTextArea
                     ref="textAreaRef"
                     @submit="handleSubmit"
@@ -24,6 +101,7 @@
                     icon="pi pi-send"
                     :label="isAwaitingInput ? 'Reply' : $tmpl._('agent_submit_button_label', 'Submit')"
                     :foo="$tmpl"
+                    :disabled="!isAwaitingInput && hasPendingUploads"
                 />
             </div>
         </div>
@@ -32,14 +110,15 @@
 
 
 <script setup lang="ts">
-import { ref, nextTick, inject, computed, watch } from "vue";
-import Card from 'primevue/card';
+import { ref, reactive, nextTick, inject, computed, watch, onBeforeUnmount } from "vue";
 import Button from 'primevue/button';
+import { filesize } from 'filesize';
 import ContainedTextArea from '../misc/ContainedTextArea.vue';
 import { useWorkflows } from '../../composables/useWorkflows';
-import { BeakerSession } from '@jataware/beaker-client';
+import { BeakerSession, type ISessionAttachment } from '@jataware/beaker-client';
 import { type BeakerSessionComponentType } from '../session/BeakerSession.vue';
 import { type BeakerNotebookComponentType } from '../notebook/BeakerNotebook.vue';
+import { uploadSessionAttachment, deleteSessionAttachment } from '../../util/attachments';
 import { marked } from "marked";
 
 const props = defineProps([
@@ -53,6 +132,28 @@ const notebook = inject<BeakerNotebookComponentType>("notebook");
 
 const query = ref("");
 const response = ref("");
+type DraftStatus = 'uploading' | 'ready' | 'error';
+type AttachmentDraft = {
+    localId: string;
+    file: File;
+    status: DraftStatus;
+    controller: AbortController;
+    attachment?: ISessionAttachment;
+    error?: string;
+};
+type ToastOptions = {
+    title: string;
+    detail: string;
+    severity: string;
+    life?: number;
+};
+
+const draftAttachments = ref<AttachmentDraft[]>([]);
+const fileInputRef = ref<HTMLInputElement>();
+const dragDepth = ref(0);
+const isDraggingFiles = computed(() => dragDepth.value > 0 && !isAwaitingInput.value);
+const hasPendingUploads = computed(() => draftAttachments.value.some((draft) => draft.status === 'uploading'));
+const showToast = inject<(options: ToastOptions) => void>("show_toast");
 defineEmits([
     "select-cell",
     "run-cell",
@@ -75,7 +176,7 @@ const inputValue = computed({
 
 const textAreaRef = ref(null);
 
-const handleSubmit = (e: any) => {
+const handleSubmit = () => {
     if (isAwaitingInput.value) {
         handleResponse();
     } else {
@@ -84,7 +185,26 @@ const handleSubmit = (e: any) => {
 }
 
 const handleQuery = () => {
-    if (!query.value.trim()) {
+    if (hasPendingUploads.value) {
+        showToast?.({
+            title: 'Attachments still uploading',
+            detail: 'Wait for the uploads to finish before sending this message.',
+            severity: 'info',
+        });
+        return;
+    }
+    if (draftAttachments.value.some((draft) => draft.status === 'error')) {
+        showToast?.({
+            title: 'Attachment upload failed',
+            detail: 'Remove failed attachments or drop them again before sending.',
+            severity: 'warn',
+        });
+        return;
+    }
+    const readyAttachments = draftAttachments.value
+        .map((draft) => draft.attachment)
+        .filter((attachment): attachment is ISessionAttachment => attachment !== undefined);
+    if (!query.value.trim() && readyAttachments.length === 0) {
         return;
     }
 
@@ -98,8 +218,10 @@ const handleQuery = () => {
             notebook.notebook.removeCell(0);
         }
     }
-    const cell = session.addQueryCell(query.value);
+    const prompt = query.value.trim() || 'Please inspect the attached file(s).';
+    const cell = session.addQueryCell(prompt, {}, readyAttachments);
     query.value = "";
+    draftAttachments.value = [];
 
     nextTick(() => {
         notebook.selectCell(cell.id);
@@ -107,6 +229,102 @@ const handleQuery = () => {
     });
 }
 
+const formatAttachmentSize = (size: number) => filesize(size, {spacer: ' '});
+
+const containsFiles = (event: DragEvent) => Array.from(event.dataTransfer?.types ?? []).includes('Files');
+
+const handleDragEnter = (event: DragEvent) => {
+    if (isAwaitingInput.value || !containsFiles(event)) return;
+    event.preventDefault();
+    dragDepth.value += 1;
+};
+
+const handleDragLeave = (event: DragEvent) => {
+    if (dragDepth.value === 0) return;
+    event.preventDefault();
+    dragDepth.value = Math.max(0, dragDepth.value - 1);
+};
+
+const handleDragOver = (event: DragEvent) => {
+    if (isAwaitingInput.value || !containsFiles(event)) return;
+    event.preventDefault();
+    if (event.dataTransfer) event.dataTransfer.dropEffect = 'copy';
+};
+
+const handleDrop = (event: DragEvent) => {
+    dragDepth.value = 0;
+    if (isAwaitingInput.value || !containsFiles(event)) return;
+    event.preventDefault();
+    void addFiles(event.dataTransfer?.files);
+};
+
+const handleFileSelection = (event: Event) => {
+    const input = event.target as HTMLInputElement;
+    void addFiles(input.files);
+    input.value = '';
+};
+
+const addFiles = async (files?: FileList | null) => {
+    if (!files?.length) return;
+    for (const file of Array.from(files)) {
+        const draft = reactive<AttachmentDraft>({
+            localId: crypto.randomUUID(),
+            file,
+            status: 'uploading',
+            controller: new AbortController(),
+        });
+        draftAttachments.value.push(draft);
+        void uploadDraft(draft);
+    }
+};
+
+const uploadDraft = async (draft: AttachmentDraft) => {
+    try {
+        if (draft.controller.signal.aborted) return;
+        draft.attachment = await uploadSessionAttachment(
+            session.sessionId,
+            draft.file,
+            draft.controller.signal,
+        );
+        draft.status = 'ready';
+        if (draft.attachment.archive_status === 'failed') {
+            showToast?.({
+                title: 'ZIP extraction failed',
+                detail: `${draft.file.name} was kept as an attachment. The agent can inspect the original ZIP. ${draft.attachment.archive_error || ''}`,
+                severity: 'warn',
+                life: 8000,
+            });
+        }
+    } catch (error) {
+        if (draft.controller.signal.aborted) return;
+        draft.status = 'error';
+        draft.error = error instanceof Error ? error.message : String(error);
+        showToast?.({
+            title: 'Attachment upload failed',
+            detail: `${draft.file.name}: ${draft.error}`,
+            severity: 'error',
+            life: 8000,
+        });
+    }
+};
+
+const removeAttachment = async (draft: AttachmentDraft) => {
+    draft.controller.abort();
+    const index = draftAttachments.value.findIndex((item) => item.localId === draft.localId);
+    if (index >= 0) draftAttachments.value.splice(index, 1);
+    if (draft.attachment) {
+        try {
+            await deleteSessionAttachment(session.sessionId, draft.attachment.id);
+        } catch (error) {
+            showToast?.({
+                title: 'Unable to remove attachment',
+                detail: error instanceof Error ? error.message : String(error),
+                severity: 'error',
+            });
+        }
+    }
+};
+
 const handleResponse = () => {
     if (!response.value.trim() || !props.awaitingInputCell) {
         return;
@@ -147,6 +365,10 @@ const placeholder = computed(() => attachedWorkflow?.value?.example_prompt ? att
 
 const questionMarkdown = computed(() => marked.parse(props?.awaitingInputQuestion ?? ""))
 
+onBeforeUnmount(() => {
+    draftAttachments.value.forEach((draft) => draft.controller.abort());
+});
+
 </script>
 
 
@@ -155,6 +377,78 @@ const questionMarkdown = computed(() => marked.parse(props?.awaitingInputQuestio
     padding: 0.5rem 0.75rem;
     background: var(--p-toolbar-background);
     border: 1px solid var(--p-toolbar-border-color);
+    position: relative;
+}
+
+.attachment-drop-overlay {
+    position: absolute;
+    inset: 0;
+    z-index: 10;
+    display: flex;
+    align-items: center;
+    justify-content: center;
+    gap: 0.6rem;
+    border: 2px dashed var(--p-primary-color);
+    border-radius: var(--p-content-border-radius);
+    background: color-mix(in srgb, var(--p-primary-color) 12%, var(--p-content-background));
+    color: var(--p-primary-color);
+    font-weight: 600;
+    pointer-events: none;
+}
+
+.attachment-drafts {
+    display: flex;
+    flex-wrap: wrap;
+    gap: 0.4rem;
+    margin: 0.4rem 0;
+}
+
+.attachment-draft {
+    display: inline-flex;
+    align-items: center;
+    gap: 0.35rem;
+    min-width: 0;
+    padding: 0.2rem 0.25rem 0.2rem 0.55rem;
+    border: 1px solid var(--p-content-border-color);
+    border-radius: 999px;
+    background: var(--p-content-background);
+    font-size: 0.85rem;
+
+    small {
+        color: var(--p-text-muted-color);
+    }
+}
+
+.attachment-name {
+    max-width: 15rem;
+    overflow: hidden;
+    text-overflow: ellipsis;
+    white-space: nowrap;
+}
+
+.attachment-error,
+.attachment-error .attachment-status-icon {
+    color: var(--p-red-500) !important;
+}
+
+.attachment-detail {
+    white-space: nowrap;
+}
+
+.attachment-remove {
+    width: 1.75rem;
+    height: 1.75rem;
+}
+
+.attachment-file-input {
+    display: none;
+}
+
+.attachment-picker {
+    flex: 0 0 auto;
+    align-self: flex-end;
+    width: 2.5rem;
+    height: 3rem;
 }
 
 #agent-prompt {
diff --git a/beaker-vue/src/components/cell/BeakerQueryCell.vue b/beaker-vue/src/components/cell/BeakerQueryCell.vue
index b9ad65b6..66ff0a09 100644
--- a/beaker-vue/src/components/cell/BeakerQueryCell.vue
+++ b/beaker-vue/src/components/cell/BeakerQueryCell.vue
@@ -10,6 +10,24 @@
                         <span class="pi pi-user query-icon"></span>
                         <span class="query-label-user">User</span>
                     </div>
+                    <div v-if="cell.session_attachments?.length" class="query-attachments">
+                        <span
+                            v-for="attachment in cell.session_attachments"
+                            :key="attachment.id"
+                            class="query-attachment"
+                            :title="attachment.archive_error || attachment.name"
+                        >
+                            <i :class="attachment.kind === 'archive' ? 'pi pi-box' : 'pi pi-paperclip'"></i>
+                            {{ attachment.name }}
+                            <small v-if="attachment.kind === 'archive' && attachment.archive_status === 'extracted'">
+                                {{ attachment.file_count }} {{ attachment.file_count === 1 ? 'file' : 'files' }}
+                            </small>
+                            <i
+                                v-if="attachment.archive_status === 'failed'"
+                                class="pi pi-exclamation-triangle attachment-warning"
+                            ></i>
+                        </span>
+                    </div>
                     <div class="query-text">{{ cell.source }}</div>
                 </div>
             </div>
@@ -218,6 +236,32 @@ onBeforeUnmount(() => {
     margin-top: 0.33rem;
 }
 
+.query-attachments {
+    display: flex;
+    flex-wrap: wrap;
+    gap: 0.4rem;
+    margin-top: 0.25rem;
+}
+
+.query-attachment {
+    display: inline-flex;
+    align-items: center;
+    gap: 0.3rem;
+    padding: 0.25rem 0.5rem;
+    border: 1px solid var(--p-content-border-color);
+    border-radius: 999px;
+    background: var(--p-content-background);
+    font-size: 0.85rem;
+
+    small {
+        color: var(--p-text-muted-color);
+    }
+}
+
+.attachment-warning {
+    color: var(--p-orange-500);
+}
+
 .collapse-control {
     display: flex;
     align-items: center;
diff --git a/beaker-vue/src/util/attachments.ts b/beaker-vue/src/util/attachments.ts
new file mode 100644
index 00000000..e863650e
--- /dev/null
+++ b/beaker-vue/src/util/attachments.ts
@@ -0,0 +1,45 @@
+import type { ISessionAttachment } from '@jataware/beaker-client';
+import { fetch } from '@/util/fetch';
+
+
+const route = (sessionId: string, attachmentId?: string) => {
+    const base = `/beaker/attachments/${encodeURIComponent(sessionId)}`;
+    return attachmentId ? `${base}/${encodeURIComponent(attachmentId)}` : base;
+};
+
+const errorMessage = async (response: Response): Promise<string> => {
+    try {
+        const payload = await response.json();
+        return payload.message || response.statusText;
+    } catch {
+        return response.statusText || `Request failed with status ${response.status}`;
+    }
+};
+
+export async function uploadSessionAttachment(
+    sessionId: string,
+    file: File,
+    signal?: AbortSignal,
+): Promise<ISessionAttachment> {
+    const form = new FormData();
+    form.append('file', file, file.name);
+    const response = await fetch(route(sessionId), {
+        method: 'POST',
+        body: form,
+        signal,
+    });
+    if (!response.ok) {
+        throw new Error(await errorMessage(response));
+    }
+    return await response.json() as ISessionAttachment;
+}
+
+export async function deleteSessionAttachment(
+    sessionId: string,
+    attachmentId: string,
+): Promise<void> {
+    const response = await fetch(route(sessionId, attachmentId), {method: 'DELETE'});
+    if (!response.ok && response.status !== 404) {
+        throw new Error(await errorMessage(response));
+    }
+}
diff --git a/beaker-vue/src/util/index.ts b/beaker-vue/src/util/index.ts
index 426ffa8a..63ecc9d7 100644
--- a/beaker-vue/src/util/index.ts
+++ b/beaker-vue/src/util/index.ts
@@ -125,3 +125,4 @@ export function atEndOfInput(input: object) {
 export * from './annotations';
 export * from './autocomplete';
 export * from './integration';
+export * from './attachments';
diff --git a/beaker-vue/tests/attachments.spec.ts b/beaker-vue/tests/attachments.spec.ts
new file mode 100644
index 00000000..20b1431d
--- /dev/null
+++ b/beaker-vue/tests/attachments.spec.ts
@@ -0,0 +1,92 @@
+import {expect, test, type Page} from '@playwright/test';
+
+
+type BrowserFile = {
+  name: string;
+  type: string;
+  bytes: number[];
+};
+
+async function dropFiles(page: Page, files: BrowserFile[]) {
+  const transfer = await page.evaluateHandle((items) => {
+    const dataTransfer = new DataTransfer();
+    for (const item of items) {
+      dataTransfer.items.add(new File([new Uint8Array(item.bytes)], item.name, {type: item.type}));
+    }
+    return dataTransfer;
+  }, files);
+  const composer = page.locator('#agent-input');
+  await composer.dispatchEvent('dragenter', {dataTransfer: transfer});
+  await expect(composer.getByText('Drop files to attach them to this message')).toBeVisible();
+  await composer.dispatchEvent('drop', {dataTransfer: transfer});
+}
+
+test('notebook chat attachments upload, remove, extract ZIPs, and send without text', async ({page}) => {
+  const baseUrl = process.env.BEAKER_E2E_URL ?? 'http://localhost:8888';
+  const sessionId = `attachment-e2e-${Date.now()}`;
+  await page.goto(`${baseUrl}/?session=${sessionId}`);
+  await expect(page.locator('span.status-label')).toHaveText('Ready', {timeout: 30_000});
+
+  await dropFiles(page, [
+    {
+      name: 'sales.csv',
+      type: 'text/csv',
+      bytes: Array.from(new TextEncoder().encode('region,revenue\nwest,42\neast,35\n')),
+    },
+    {
+      name: 'notes.txt',
+      type: 'text/plain',
+      bytes: Array.from(new TextEncoder().encode('temporary notes')),
+    },
+  ]);
+
+  const drafts = page.getByTestId('attachment-draft');
+  await expect(drafts).toHaveCount(2);
+  await expect(drafts.filter({hasText: 'sales.csv'}).locator('.pi-spinner')).toHaveCount(0, {timeout: 15_000});
+  await expect(drafts.filter({hasText: 'notes.txt'}).locator('.pi-spinner')).toHaveCount(0, {timeout: 15_000});
+
+  const deleteRequest = page.waitForResponse(
+    (response) => response.request().method() === 'DELETE' && response.url().includes('/beaker/attachments/'),
+  );
+  await drafts.filter({hasText: 'notes.txt'}).getByRole('button', {name: 'Remove notes.txt'}).click();
+  expect((await deleteRequest).status()).toBe(204);
+  await expect(drafts).toHaveCount(1);
+  await expect(drafts).toContainText('sales.csv');
+
+  const zipBytes = Uint8Array.from(atob(
+    'UEsDBBQAAAAIANpc7lxygqJGHwAAAB8AAAAOAAAAZGF0YS9zYWxlcy5jc3YrSk3PzM/TKUotS80rTeUqTy0u0TEx4kpNBNLGplwAUEsDBBQAAAAIANpc7lx6D5rSEAAAAA4AAAAKAAAAUkVBRE1FLnR4dCtOzC3ISVVISSxJLE4tAQBQSwECFAMUAAAACADaXO5ccoKiRh8AAAAfAAAADgAAAAAAAAAAAAAAgAEAAAAAZGF0YS9zYWxlcy5jc3ZQSwECFAMUAAAACADaXO5ceg+a0hAAAAAOAAAACgAAAAAAAAAAAAAAgAFLAAAAUkVBRE1FLnR4dFBLBQYAAAAAAgACAHQAAACDAAAAAAA=',
+  ), (character) => character.charCodeAt(0));
+  await dropFiles(page, [{name: 'dataset.zip', type: 'application/zip', bytes: Array.from(zipBytes)}]);
+  await expect(drafts).toHaveCount(2);
+  await expect(drafts.filter({hasText: 'dataset.zip'})).toContainText('2 files', {timeout: 15_000});
+
+  await drafts.filter({hasText: 'sales.csv'}).getByRole('button', {name: 'Remove sales.csv'}).click();
+  await expect(drafts).toHaveCount(1);
+  await expect(drafts).toContainText('dataset.zip');
+
+  await page.getByRole('button', {name: 'Submit'}).click();
+  await expect(drafts).toHaveCount(0);
+  const queryCell = page.locator('.next-query-cell').last();
+  await expect(queryCell).toContainText('dataset.zip');
+  await expect(queryCell).toContainText('Please inspect the attached file(s).');
+
+  await expect.poll(async () => {
+    const response = await page.request.get(`${baseUrl}/beaker/attachments/${sessionId}`);
+    if (!response.ok()) return [];
+    const attachments = await response.json();
+    return attachments.map((attachment: {name: string; committed: boolean; archive_status?: string}) => ({
+      name: attachment.name,
+      committed: attachment.committed,
+      archiveStatus: attachment.archive_status,
+    }));
+  }).toEqual([{name: 'dataset.zip', committed: true, archiveStatus: 'extracted'}]);
+
+  const xsrfCookie = (await page.context().cookies()).find((cookie) => cookie.name === '_xsrf');
+  expect(xsrfCookie).toBeDefined();
+  const clearResponse = await page.request.delete(`${baseUrl}/beaker/attachments/${sessionId}`, {
+    headers: {'X-XSRFToken': xsrfCookie!.value},
+  });
+  expect(clearResponse.status()).toBe(204);
+  const remainingResponse = await page.request.get(`${baseUrl}/beaker/attachments/${sessionId}`);
+  expect(await remainingResponse.json()).toEqual([]);
+});
diff --git a/src/beaker_notebook/app/api/attachments.py b/src/beaker_notebook/app/api/attachments.py
new file mode 100644
index 00000000..4b31633b
--- /dev/null
+++ b/src/beaker_notebook/app/api/attachments.py
@@ -0,0 +1,143 @@
+from __future__ import annotations
+
+import asyncio
+import json
+from typing import Any
+
+from jupyter_server.base.handlers import JupyterHandler
+from tornado import web
+
+from beaker_notebook.services.attachments import AttachmentError, SessionAttachmentManager
+from beaker_notebook.services.auth import BeakerUser
+
+
+class SessionAttachmentHandler(JupyterHandler):
+    """Upload, list, inspect, and remove temporary chat attachments."""
+
+    def set_default_headers(self):
+        self.set_header("Content-Type", "application/json")
+
+    def write_error(self, status_code: int, **kwargs):
+        message = self._reason
+        exception = kwargs.get("exc_info", (None, None, None))[1]
+        if isinstance(exception, web.HTTPError) and exception.log_message:
+            message = exception.log_message
+        self.finish(json.dumps({"status": status_code, "message": message}))
+
+    @property
+    def attachment_manager(self) -> SessionAttachmentManager:
+        manager = getattr(self.serverapp, "attachment_manager", None)
+        if manager is None:
+            raise web.HTTPError(503, "Session attachment storage is unavailable")
+        return manager
+
+    def _authenticated_request_kernel_id(self) -> str | None:
+        header = self.request.headers.get(self.identity_provider.beaker_kernel_header)
+        if not header:
+            return None
+
+        kernel_id = self.identity_provider.authenticated_beaker_kernel_id(self)
+        if kernel_id is None:
+            raise web.HTTPError(403, "Invalid kernel authentication")
+        return kernel_id
+
+    async def _resolve_session(self, session_id: str) -> tuple[dict[str, Any], Any, str | None]:
+        if not session_id or session_id in (".", "..") or "\x00" in session_id:
+            raise web.HTTPError(400, "Invalid session ID")
+        try:
+            session = await self.session_manager.get_session(path=session_id)
+        except web.HTTPError as err:
+            if err.status_code == 404:
+                raise web.HTTPError(404, "Session not found") from err
+            raise
+
+        kernel_id = session.get("kernel", {}).get("id") if session.get("kernel") else session.get("kernel_id")
+        kernel_manager = self.kernel_manager.get_kernel(kernel_id) if kernel_id else None
+        owner = getattr(kernel_manager, "user", None)
+
+        request_kernel_id = self._authenticated_request_kernel_id()
+        if request_kernel_id:
+            if request_kernel_id != kernel_id:
+                raise web.HTTPError(403, "Kernel does not own this session")
+            return session, owner, request_kernel_id
+
+        current = self.current_user
+        if isinstance(owner, BeakerUser) and isinstance(current, BeakerUser):
+            if owner.username != current.username:
+                raise web.HTTPError(403, "Session does not belong to the current user")
+        return session, owner or current, None
+
+    def _write_json(self, value: Any) -> None:
+        self.write(json.dumps(value))
+
+    async def get(self, session_id: str, attachment_id: str | None = None):
+        _, owner, request_kernel_id = await self._resolve_session(session_id)
+        try:
+            if attachment_id:
+                result = await asyncio.to_thread(
+                    self.attachment_manager.get_attachment, session_id, attachment_id, owner
+                )
+            else:
+                current_ids = self.get_arguments("current") if request_kernel_id else []
+                if request_kernel_id:
+                    result = await asyncio.to_thread(
+                        self.attachment_manager.commit_attachments,
+                        session_id,
+                        current_ids,
+                        owner,
+                    )
+                else:
+                    result = await asyncio.to_thread(
+                        self.attachment_manager.list_attachments, session_id, owner
+                    )
+        except AttachmentError as err:
+            raise web.HTTPError(404, str(err)) from err
+        self._write_json(result)
+
+    async def post(self, session_id: str, attachment_id: str | None = None):
+        if attachment_id:
+            raise web.HTTPError(405)
+        _, owner, _ = await self._resolve_session(session_id)
+        uploads = self.request.files.get("file", [])
+        if len(uploads) != 1:
+            raise web.HTTPError(400, "Upload exactly one file per request")
+        upload = uploads[0]
+        try:
+            result = await asyncio.to_thread(
+                self.attachment_manager.create_attachment,
+                session_id,
+                upload.get("filename", "upload"),
+                upload.get("content_type"),
+                upload.get("body", b""),
+                owner,
+            )
+        except AttachmentError as err:
+            raise web.HTTPError(400, str(err)) from err
+        self.set_status(201)
+        self._write_json(result)
+
+    async def delete(self, session_id: str, attachment_id: str | None = None):
+        _, owner, _ = await self._resolve_session(session_id)
+        if not attachment_id:
+            await asyncio.to_thread(
+                self.attachment_manager.delete_session, session_id, owner
+            )
+            self.set_status(204)
+            self.finish()
+            return
+        try:
+            await asyncio.to_thread(
+                self.attachment_manager.delete_attachment, session_id, attachment_id, owner
+            )
+        except AttachmentError as err:
+            raise web.HTTPError(404, str(err)) from err
+        self.set_status(204)
+        self.finish()
+
+
+handlers = [
+    (
+        r"attachments/(?P<session_id>[^/]+)/?(?P<attachment_id>[0-9a-fA-F-]+)?",
+        SessionAttachmentHandler,
+    ),
+]
diff --git a/src/beaker_notebook/app/base.py b/src/beaker_notebook/app/base.py
index 66c8d07f..93dfaed8 100644
--- a/src/beaker_notebook/app/base.py
+++ b/src/beaker_notebook/app/base.py
@@ -158,6 +158,8 @@ def init_configurables(self, **kwargs):
         self.notebook_manager = self.notebook_manager_class(
             parent=self,
         )
+        from beaker_notebook.services.attachments import SessionAttachmentManager
+        self.attachment_manager = SessionAttachmentManager(parent=self)
         from beaker_notebook.services.context.manager import BeakerContextManager
         self.context_manager: BeakerContextManager = self.context_manager_class(parent=self)
 
@@ -272,8 +274,21 @@ def _default_virtual_home_root(self):
     def _default_root_dir(self):
         return self.working_dir or super()._default_root_dir()
 
+    @property
+    def local_mode(self) -> bool:
+        """Whether this app runs as a single-user local notebook rather than a multi-user server.
+
+        Server deployments keep per-user state namespaced under each user's virtual home;
+        local deployments centralize it under the shared Beaker data directory. Subclasses
+        that represent a local launch (e.g. ``BeakerNotebookApp``) override this to ``True``.
+        """
+        return False
+
     def stop(self, from_signal = False, **kwargs):
         print("Shutting down Beaker server...")
+        attachment_manager = getattr(self, "attachment_manager", None)
+        if attachment_manager is not None:
+            attachment_manager.cleanup()
         return super().stop(from_signal, **kwargs)
 
     @property
diff --git a/src/beaker_notebook/app/notebook_app.py b/src/beaker_notebook/app/notebook_app.py
index 26d2a3b9..266c1ca5 100644
--- a/src/beaker_notebook/app/notebook_app.py
+++ b/src/beaker_notebook/app/notebook_app.py
@@ -18,5 +18,9 @@ class BeakerNotebookApp(BaseBeakerApp):
         }
     }
 
+    @property
+    def local_mode(self) -> bool:
+        return True
+
 if __name__ == "__main__":
     BeakerNotebookApp.launch_instance()
diff --git a/src/beaker_notebook/kernel.py b/src/beaker_notebook/kernel.py
index 201e448e..ce30d985 100644
--- a/src/beaker_notebook/kernel.py
+++ b/src/beaker_notebook/kernel.py
@@ -9,6 +9,7 @@
 import sys
 import traceback
 import uuid
+import urllib.parse
 from functools import partial
 from typing import Literal, Optional, ClassVar, Awaitable
 
@@ -174,6 +175,52 @@ def api_auth(self) -> str:
 
         return f"{preamble}:{kernel_id}:{nonce}:{hash_value}"
 
+    def get_session_attachments(self, current_attachment_ids: list[str] | None = None) -> list[dict]:
+        """Fetch attachment metadata after the server validates kernel/session ownership."""
+        current_ids = set(current_attachment_ids or [])
+        session_id = self.beaker_session or self.session_id
+        url = url_path_join(
+            self.jupyter_server,
+            "/beaker/attachments/",
+            urllib.parse.quote(str(session_id), safe=""),
+        )
+        response = requests.get(
+            url,
+            params=[("current", attachment_id) for attachment_id in current_ids],
+            headers={"X-AUTH-BEAKER": self.api_auth()},
+            timeout=10,
+        )
+        if response.status_code >= 400:
+            raise ValueError(
+                f"Unable to load session attachments (status {response.status_code}): {response.text}"
+            )
+        attachments = response.json()
+        available_ids = {item.get("id") for item in attachments}
+        missing = current_ids - available_ids
+        if missing:
+            raise ValueError(f"Session attachments are no longer available: {sorted(missing)}")
+        for item in attachments:
+            item["current"] = item.get("id") in current_ids
+        return attachments
+
+    def clear_session_attachments(self) -> None:
+        """Delete every temporary attachment owned by the current notebook session."""
+        session_id = self.beaker_session or self.session_id
+        url = url_path_join(
+            self.jupyter_server,
+            "/beaker/attachments/",
+            urllib.parse.quote(str(session_id), safe=""),
+        )
+        response = requests.delete(
+            url,
+            headers={"X-AUTH-BEAKER": self.api_auth()},
+            timeout=10,
+        )
+        if response.status_code >= 400:
+            raise ValueError(
+                f"Unable to clear session attachments (status {response.status_code}): {response.text}"
+            )
+
     async def handle_magic_word(self, server, target_stream, data):
         message = JupyterMessage.parse(data)
         cell_content: str = message.content.get("code", "").strip()
@@ -603,9 +650,16 @@ async def llm_request(self, message: JupyterMessage):
         request = content.get("request", None)
         if not request:
             return
+
+        attachment_ids = content.get("attachments", []) or []
+        if not isinstance(attachment_ids, list) or not all(isinstance(item, str) for item in attachment_ids):
+            raise ValueError("attachments must be a list of attachment IDs")
+        session_attachments = self.get_session_attachments(attachment_ids)
+
         if not self.context:
             raise Exception("Context has not been set")
         setattr(self.context, "current_llm_query", request)
+        self.context.session_attachments = session_attachments
 
         notebook_state = message.metadata.get("notebook_state", None)
         if notebook_state is not None:
@@ -805,6 +859,8 @@ async def set_agent_model(self, message):
     @message_handler
     async def reset_kernel(self, message):
         reset_config()
+        self.clear_session_attachments()
+        self.context.session_attachments = []
         await self.set_context(
             self.context.SLUG,
             self.context.config,
diff --git a/src/beaker_notebook/lib/agent.py b/src/beaker_notebook/lib/agent.py
index f1d6dbb6..a3f545a9 100644
--- a/src/beaker_notebook/lib/agent.py
+++ b/src/beaker_notebook/lib/agent.py
@@ -1,6 +1,7 @@
 import json
 import logging
 import mimetypes
+from pathlib import Path
 import typing
 import urllib.parse
 
@@ -304,6 +305,45 @@ async def ask_user(
     def _send_notebook_state(self) -> bool:
         return bool(config.send_notebook_state and self.context and self.context.notebook_state)
 
+    def _has_session_attachments(self) -> bool:
+        return bool(self.context and self.context.session_attachments)
+
+    @statetool(condition=_has_session_attachments, name="session_attachments")
+    async def session_attachments(self) -> str:
+        """Provides the temporary files available in the current notebook session.
+
+        Returns:
+            str: JSON attachment metadata and exact paths suitable for get_session_attachment or run_code.
+        """
+        attachments = []
+        for item in self.context.session_attachments:
+            files = item.get("files", [])
+            attachments.append({
+                "id": item.get("id"),
+                "name": item.get("name"),
+                "mimetype": item.get("mimetype"),
+                "size": item.get("size"),
+                "kind": item.get("kind"),
+                "current": bool(item.get("current")),
+                "path": item.get("path"),
+                "root_path": item.get("root_path"),
+                "original_path": item.get("original_path"),
+                "archive_status": item.get("archive_status"),
+                "archive_error": item.get("archive_error"),
+                "file_count": item.get("file_count", len(files)),
+                "files": files[:100],
+                "manifest_truncated": len(files) > 100,
+            })
+        return json.dumps({
+            "instructions": (
+                "These are temporary session attachments supplied by the user. "
+                "Items with current=true were attached to the current request. Use the exact path with run_code; "
+                "for extracted archives, files are beneath root_path. Use get_session_attachment when direct "
+                "text or multimedia inspection is more appropriate. Do not claim a file is permanent."
+            ),
+            "attachments": attachments,
+        })
+
 
     @statetool(condition=_send_notebook_state, name="notebook_state")
     async def notebook_state(self) -> str:
@@ -345,6 +385,59 @@ async def get_notebook_cell(self, cell_id: str) -> dict:
         cell = next((cell for cell in nbstate["cells"] if cell["id"] == cell_id), None)
         return format_cell(cell, truncate_content=False, truncate_outputs=False, exclude_media=True)
 
+    @tool(autosummarize=True, summarizer=succinct_tool_summarizer(), internal=True)
+    async def get_session_attachment(self, attachment_id: str, relative_path: str | None = None) -> MultiModalResponse | str:
+        """Reads a temporary session attachment or a file within an extracted ZIP.
+
+        Args:
+            attachment_id: The attachment ID supplied by the session_attachments state.
+            relative_path: For an extracted archive, an optional path beneath its root_path.
+
+        Returns:
+            Multimedia content for image/audio/video files, a bounded preview for text files, or file metadata
+            instructing the agent to use run_code for structured or binary processing.
+        """
+        attachment = next(
+            (item for item in self.context.session_attachments if item.get("id") == attachment_id),
+            None,
+        )
+        if attachment is None:
+            raise ValueError(f"Unknown session attachment ID: {attachment_id!r}")
+
+        if relative_path:
+            root = Path(attachment["root_path"]).resolve()
+            target = (root / relative_path).resolve()
+            if not target.is_relative_to(root):
+                raise ValueError("relative_path must remain inside the extracted attachment directory")
+        else:
+            path = attachment.get("path") or attachment.get("original_path")
+            if not path:
+                return json.dumps(attachment)
+            target = Path(path).resolve()
+
+        if not target.is_file():
+            raise ValueError(f"Attachment file not found: {target}")
+        mimetype = mimetypes.guess_type(target.name)[0] or attachment.get("mimetype") or "application/octet-stream"
+        if is_multimedia_mimetype(mimetype):
+            return MultiModalResponse.from_file(target, mimetype=mimetype)
+
+        text_suffixes = {".csv", ".tsv", ".json", ".jsonl", ".md", ".txt", ".log", ".xml", ".yaml", ".yml"}
+        if mimetype.startswith("text/") or target.suffix.lower() in text_suffixes:
+            preview_limit = 50_000
+            text = target.read_text(errors="replace")
+            truncated = len(text) > preview_limit
+            preview = text[:preview_limit]
+            if truncated:
+                preview += f"\n\n[Preview truncated; use run_code with {str(target)!r} for the complete file.]"
+            return preview
+
+        return json.dumps({
+            "attachment": attachment,
+            "selected_path": str(target),
+            "mimetype": mimetype,
+            "instructions": "Use run_code with selected_path to inspect or process this binary/structured file.",
+        })
+
 
     @tool(autosummarize=True, summarizer=succinct_tool_summarizer(), internal=True)
     async def get_notebook_multimedia_output(self, refs: list[str]) -> MultiModalResponse:
diff --git a/src/beaker_notebook/lib/config.py b/src/beaker_notebook/lib/config.py
index d854264f..3b477fdf 100644
--- a/src/beaker_notebook/lib/config.py
+++ b/src/beaker_notebook/lib/config.py
@@ -294,6 +294,41 @@ class ConfigClass:
         sensitive=False,
         label="describe_variables sample budget (0 = use subkernel default)"
     )
+    attachment_max_upload_bytes: int = configfield(
+        description="Maximum size, in bytes, of one temporary chat attachment.",
+        env_var="ATTACHMENT_MAX_UPLOAD_BYTES",
+        default=25 * 1024 * 1024,
+        normalize_function=int,
+        label="Maximum chat attachment size (bytes)",
+    )
+    attachment_max_session_bytes: int = configfield(
+        description="Maximum combined stored size, in bytes, of temporary chat attachments in one session.",
+        env_var="ATTACHMENT_MAX_SESSION_BYTES",
+        default=250 * 1024 * 1024,
+        normalize_function=int,
+        label="Maximum session attachment storage (bytes)",
+    )
+    attachment_max_extracted_bytes: int = configfield(
+        description="Maximum total uncompressed size, in bytes, extracted from one ZIP chat attachment.",
+        env_var="ATTACHMENT_MAX_EXTRACTED_BYTES",
+        default=100 * 1024 * 1024,
+        normalize_function=int,
+        label="Maximum extracted ZIP size (bytes)",
+    )
+    attachment_max_archive_entries: int = configfield(
+        description="Maximum number of files that may be extracted from one ZIP chat attachment.",
+        env_var="ATTACHMENT_MAX_ARCHIVE_ENTRIES",
+        default=1000,
+        normalize_function=int,
+        label="Maximum ZIP entry count",
+    )
+    attachment_max_archive_ratio: int = configfield(
+        description="Maximum allowed uncompressed-to-compressed size ratio for automatic ZIP extraction.",
+        env_var="ATTACHMENT_MAX_ARCHIVE_RATIO",
+        default=100,
+        normalize_function=int,
+        label="Maximum ZIP compression ratio",
+    )
 
     @property
     def checkpoint_storage_path(self):
diff --git a/src/beaker_notebook/lib/context.py b/src/beaker_notebook/lib/context.py
index 9e2065dd..6a7d9539 100644
--- a/src/beaker_notebook/lib/context.py
+++ b/src/beaker_notebook/lib/context.py
@@ -122,6 +122,7 @@ def __init__(
         )
 
         self.current_llm_query = None
+        self.session_attachments: list[dict[str, Any]] = []
         self._notebook_state = None
 
         self.disable_tools()
diff --git a/src/beaker_notebook/services/attachments.py b/src/beaker_notebook/services/attachments.py
new file mode 100644
index 00000000..6670d4ef
--- /dev/null
+++ b/src/beaker_notebook/services/attachments.py
@@ -0,0 +1,365 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import mimetypes
+import shutil
+import stat
+import threading
+import uuid
+import zipfile
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path, PurePosixPath
+from typing import Any
+
+from beaker_notebook.lib.config import config
+from beaker_notebook.services.auth import BeakerUser
+from beaker_notebook.services.storage import BEAKER_LOCAL_DATA_PATH
+
+
+class AttachmentError(ValueError):
+    """A user-facing attachment validation or storage error."""
+
+
+@dataclass(frozen=True)
+class AttachmentLimits:
+    max_upload_bytes: int = 25 * 1024 * 1024
+    max_session_bytes: int = 250 * 1024 * 1024
+    max_extracted_bytes: int = 100 * 1024 * 1024
+    max_archive_entries: int = 1000
+    max_archive_ratio: int = 100
+
+    @classmethod
+    def from_config(cls) -> "AttachmentLimits":
+        return cls(
+            max_upload_bytes=int(config.attachment_max_upload_bytes),
+            max_session_bytes=int(config.attachment_max_session_bytes),
+            max_extracted_bytes=int(config.attachment_max_extracted_bytes),
+            max_archive_entries=int(config.attachment_max_archive_entries),
+            max_archive_ratio=int(config.attachment_max_archive_ratio),
+        )
+
+
+class SessionAttachmentManager:
+    """Stores temporary chat attachments in a hidden, session-scoped directory."""
+
+    METADATA_FILENAME = "attachment.json"
+
+    def __init__(self, parent: Any, limits: AttachmentLimits | None = None):
+        self.parent = parent
+        self.limits = limits or AttachmentLimits.from_config()
+        self.instance_id = uuid.uuid4().hex
+        self._instance_roots: set[Path] = set()
+        self._mutation_lock = threading.RLock()
+
+    @staticmethod
+    def _safe_filename(filename: str) -> str:
+        filename = (filename or "upload").replace("\\", "/")
+        basename = PurePosixPath(filename).name.strip().replace("\x00", "")
+        if basename in ("", ".", ".."):
+            return "upload"
+        return basename
+
+    @staticmethod
+    def _session_key(session_id: str) -> str:
+        return hashlib.sha256(session_id.encode("utf-8")).hexdigest()[:24]
+
+    def _user_root(self, user: Any = None) -> Path:
+        if isinstance(user, BeakerUser) and user.home_dir:
+            return (Path(self.parent.virtual_home_root) / user.home_dir).resolve()
+        return Path(self.parent.root_dir).resolve()
+
+    def _attachments_base(self, user: Any = None) -> Path:
+        """Root under which this runtime stores session attachments.
+
+        In local mode all attachments live beneath the shared Beaker data directory,
+        independent of the requesting identity, so a ``beaker notebook`` launched in any
+        working directory keeps its temporary files in one predictable place. In server
+        mode they are namespaced under the owning user's virtual home to preserve per-user
+        isolation.
+        """
+        if getattr(self.parent, "local_mode", False):
+            return Path(BEAKER_LOCAL_DATA_PATH).resolve() / "session-attachments"
+        return self._user_root(user) / ".beaker" / "session-attachments"
+
+    def _session_root(self, session_id: str, user: Any = None) -> Path:
+        instance_root = self._attachments_base(user) / self.instance_id
+        self._instance_roots.add(instance_root)
+        return instance_root / self._session_key(session_id)
+
+    def _attachment_root(self, session_id: str, attachment_id: str, user: Any = None) -> Path:
+        try:
+            normalized_id = str(uuid.UUID(attachment_id))
+        except (ValueError, TypeError, AttributeError) as err:
+            raise AttachmentError("Invalid attachment ID") from err
+        return self._session_root(session_id, user) / normalized_id
+
+    @staticmethod
+    def _metadata_path(attachment_root: Path) -> Path:
+        return attachment_root / SessionAttachmentManager.METADATA_FILENAME
+
+    def _read_metadata(self, attachment_root: Path) -> dict[str, Any]:
+        try:
+            return json.loads(self._metadata_path(attachment_root).read_text())
+        except (FileNotFoundError, json.JSONDecodeError) as err:
+            raise AttachmentError("Attachment not found") from err
+
+    def list_attachments(self, session_id: str, user: Any = None) -> list[dict[str, Any]]:
+        session_root = self._session_root(session_id, user)
+        if not session_root.is_dir():
+            return []
+        attachments: list[dict[str, Any]] = []
+        for child in session_root.iterdir():
+            if not child.is_dir():
+                continue
+            try:
+                metadata = self._read_metadata(child)
+            except AttachmentError:
+                continue
+            if metadata.get("session_id") == session_id:
+                attachments.append(metadata)
+        attachments.sort(key=lambda item: item.get("created_at", ""))
+        return attachments
+
+    def get_attachment(self, session_id: str, attachment_id: str, user: Any = None) -> dict[str, Any]:
+        root = self._attachment_root(session_id, attachment_id, user)
+        metadata = self._read_metadata(root)
+        if metadata.get("session_id") != session_id:
+            raise AttachmentError("Attachment not found")
+        return metadata
+
+    def commit_attachments(
+        self, session_id: str, attachment_ids: list[str], user: Any = None
+    ) -> list[dict[str, Any]]:
+        """Mark attachments as sent, then return all sent attachments for the session."""
+        with self._mutation_lock:
+            return self._commit_attachments(session_id, attachment_ids, user)
+
+    def _commit_attachments(
+        self, session_id: str, attachment_ids: list[str], user: Any = None
+    ) -> list[dict[str, Any]]:
+        requested = set(attachment_ids)
+        attachments = self.list_attachments(session_id, user)
+        available = {item.get("id") for item in attachments}
+        missing = requested - available
+        if missing:
+            raise AttachmentError(f"Attachments are no longer available: {sorted(missing)}")
+
+        for metadata in attachments:
+            if metadata.get("id") not in requested or metadata.get("committed"):
+                continue
+            metadata["committed"] = True
+            root = self._attachment_root(session_id, metadata["id"], user)
+            self._metadata_path(root).write_text(json.dumps(metadata, indent=2))
+        return [item for item in attachments if item.get("committed") or item.get("id") in requested]
+
+    def _session_stored_size(self, session_id: str, user: Any = None) -> int:
+        return sum(int(item.get("stored_size", item.get("size", 0))) for item in self.list_attachments(session_id, user))
+
+    @staticmethod
+    def _is_zip(filename: str, mimetype: str) -> bool:
+        return filename.lower().endswith(".zip") or mimetype in {
+            "application/zip",
+            "application/x-zip-compressed",
+        }
+
+    def _validate_zip_members(self, archive: zipfile.ZipFile) -> tuple[list[zipfile.ZipInfo], int]:
+        files: list[zipfile.ZipInfo] = []
+        seen_paths: set[str] = set()
+        total_size = 0
+        total_compressed = 0
+
+        for info in archive.infolist():
+            raw_name = info.filename.replace("\\", "/")
+            path = PurePosixPath(raw_name)
+            if not raw_name or raw_name.startswith("/") or path.is_absolute() or ".." in path.parts:
+                raise AttachmentError(f"ZIP contains an unsafe path: {info.filename!r}")
+            normalized = path.as_posix().rstrip("/")
+            if not normalized:
+                continue
+            if normalized in seen_paths:
+                raise AttachmentError(f"ZIP contains a duplicate path: {normalized!r}")
+            seen_paths.add(normalized)
+
+            mode = info.external_attr >> 16
+            if stat.S_ISLNK(mode):
+                raise AttachmentError(f"ZIP contains a symbolic link: {normalized!r}")
+            if info.flag_bits & 0x1:
+                raise AttachmentError("Encrypted ZIP files are not extracted automatically")
+            if info.is_dir():
+                continue
+            if stat.S_IFMT(mode) and not stat.S_ISREG(mode):
+                raise AttachmentError(f"ZIP contains a special file: {normalized!r}")
+
+            files.append(info)
+            total_size += int(info.file_size)
+            total_compressed += int(info.compress_size)
+
+        if len(files) > self.limits.max_archive_entries:
+            raise AttachmentError(
+                f"ZIP contains {len(files)} files; the limit is {self.limits.max_archive_entries}"
+            )
+        if total_size > self.limits.max_extracted_bytes:
+            raise AttachmentError(
+                f"ZIP expands to {total_size} bytes; the limit is {self.limits.max_extracted_bytes}"
+            )
+        ratio = total_size / max(total_compressed, 1)
+        if ratio > self.limits.max_archive_ratio:
+            raise AttachmentError(
+                f"ZIP compression ratio is {ratio:.1f}:1; the limit is {self.limits.max_archive_ratio}:1"
+            )
+        return files, total_size
+
+    def _extract_zip(self, archive_path: Path, destination: Path) -> tuple[list[str], int]:
+        temporary = destination.with_name(f"{destination.name}.extracting")
+        shutil.rmtree(temporary, ignore_errors=True)
+        temporary.mkdir(parents=True)
+        try:
+            with zipfile.ZipFile(archive_path) as archive:
+                members, expected_size = self._validate_zip_members(archive)
+                extracted_size = 0
+                manifest: list[str] = []
+                for info in members:
+                    relative = PurePosixPath(info.filename.replace("\\", "/"))
+                    target = temporary.joinpath(*relative.parts)
+                    target.parent.mkdir(parents=True, exist_ok=True)
+                    with archive.open(info) as source, target.open("xb") as output:
+                        while chunk := source.read(1024 * 1024):
+                            extracted_size += len(chunk)
+                            if extracted_size > self.limits.max_extracted_bytes:
+                                raise AttachmentError("ZIP exceeded the extracted-size limit while unpacking")
+                            output.write(chunk)
+                    manifest.append(relative.as_posix())
+                if extracted_size != expected_size:
+                    raise AttachmentError("ZIP contents did not match the declared uncompressed size")
+            destination.parent.mkdir(parents=True, exist_ok=True)
+            temporary.replace(destination)
+            return manifest, extracted_size
+        except Exception:
+            shutil.rmtree(temporary, ignore_errors=True)
+            raise
+
+    def create_attachment(
+        self,
+        session_id: str,
+        filename: str,
+        mimetype: str | None,
+        body: bytes,
+        user: Any = None,
+    ) -> dict[str, Any]:
+        with self._mutation_lock:
+            return self._create_attachment(session_id, filename, mimetype, body, user)
+
+    def _create_attachment(
+        self,
+        session_id: str,
+        filename: str,
+        mimetype: str | None,
+        body: bytes,
+        user: Any = None,
+    ) -> dict[str, Any]:
+        if not body:
+            raise AttachmentError("The uploaded file is empty")
+        if len(body) > self.limits.max_upload_bytes:
+            raise AttachmentError(
+                f"File is {len(body)} bytes; the per-file limit is {self.limits.max_upload_bytes}"
+            )
+        if self._session_stored_size(session_id, user) + len(body) > self.limits.max_session_bytes:
+            raise AttachmentError("This upload would exceed the session attachment storage limit")
+
+        attachment_id = str(uuid.uuid4())
+        safe_name = self._safe_filename(filename)
+        resolved_mimetype = mimetype or mimetypes.guess_type(safe_name)[0] or "application/octet-stream"
+        attachment_root = self._attachment_root(session_id, attachment_id, user)
+        attachment_root.mkdir(parents=True, exist_ok=False)
+
+        is_zip = self._is_zip(safe_name, resolved_mimetype)
+        original_path: Path | None = None
+        file_path: Path | None = None
+        root_path = attachment_root / "files"
+        archive_status: str | None = None
+        archive_error: str | None = None
+        manifest: list[str] = []
+        extracted_size = 0
+
+        try:
+            if is_zip:
+                original_path = attachment_root / "original" / safe_name
+                original_path.parent.mkdir(parents=True)
+                original_path.write_bytes(body)
+                archive_status = "extracted"
+                try:
+                    manifest, extracted_size = self._extract_zip(original_path, root_path)
+                    if self._session_stored_size(session_id, user) + len(body) + extracted_size > self.limits.max_session_bytes:
+                        raise AttachmentError("Extracted ZIP would exceed the session attachment storage limit")
+                except Exception as err:
+                    shutil.rmtree(root_path, ignore_errors=True)
+                    root_path.mkdir(parents=True, exist_ok=True)
+                    archive_status = "failed"
+                    archive_error = str(err)
+                    manifest = []
+                    extracted_size = 0
+            else:
+                root_path.mkdir(parents=True)
+                file_path = root_path / safe_name
+                file_path.write_bytes(body)
+                manifest = [safe_name]
+
+            stored_size = len(body) + extracted_size
+            metadata = {
+                "id": attachment_id,
+                "session_id": session_id,
+                "name": safe_name,
+                "mimetype": resolved_mimetype,
+                "size": len(body),
+                "stored_size": stored_size,
+                "created_at": datetime.now(timezone.utc).isoformat(),
+                "committed": False,
+                "kind": "archive" if is_zip else "file",
+                "path": str(file_path.resolve()) if file_path else None,
+                "root_path": str(root_path.resolve()),
+                "original_path": str(original_path.resolve()) if original_path else None,
+                "archive_status": archive_status,
+                "archive_error": archive_error,
+                "file_count": len(manifest),
+                "files": manifest,
+            }
+            self._metadata_path(attachment_root).write_text(json.dumps(metadata, indent=2))
+            return metadata
+        except Exception:
+            shutil.rmtree(attachment_root, ignore_errors=True)
+            raise
+
+    def delete_attachment(self, session_id: str, attachment_id: str, user: Any = None) -> None:
+        with self._mutation_lock:
+            self._delete_attachment(session_id, attachment_id, user)
+
+    def _delete_attachment(self, session_id: str, attachment_id: str, user: Any = None) -> None:
+        attachment_root = self._attachment_root(session_id, attachment_id, user)
+        metadata = self._read_metadata(attachment_root)
+        if metadata.get("session_id") != session_id:
+            raise AttachmentError("Attachment not found")
+        shutil.rmtree(attachment_root)
+
+    def delete_session(self, session_id: str, user: Any = None) -> None:
+        with self._mutation_lock:
+            shutil.rmtree(self._session_root(session_id, user), ignore_errors=True)
+
+    def cleanup(self) -> None:
+        """Remove every temporary attachment directory created by this server runtime."""
+        with self._mutation_lock:
+            session_attachment_dirs = {instance_root.parent for instance_root in self._instance_roots}
+            for instance_root in self._instance_roots:
+                shutil.rmtree(instance_root, ignore_errors=True)
+            self._instance_roots.clear()
+            for session_attachment_dir in session_attachment_dirs:
+                # Prune the shared "session-attachments" directory only if this was the last
+                # runtime using it. Never climb higher: in local mode its parent is the shared
+                # Beaker data directory, and in server mode it is the user's ".beaker" — neither
+                # is ours to remove.
+                try:
+                    session_attachment_dir.rmdir()
+                except OSError:
+                    # Another server runtime still owns it, or it is not empty.
+                    pass
diff --git a/src/beaker_notebook/services/auth/__init__.py b/src/beaker_notebook/services/auth/__init__.py
index d9dda83c..aa69ee36 100644
--- a/src/beaker_notebook/services/auth/__init__.py
+++ b/src/beaker_notebook/services/auth/__init__.py
@@ -1,15 +1,16 @@
 import contextvars
-import inspect
-import os
 import hashlib
+import hmac
+import inspect
 import logging
+import os
 from dataclasses import dataclass, field
 from functools import lru_cache, update_wrapper, wraps
-from traitlets import Unicode, Bool
 from typing import Optional
 
 from jupyter_server.auth.authorizer import Authorizer
 from jupyter_server.auth.identity import IdentityProvider, User
+from traitlets import Bool, Unicode
 from tornado import web
 
 from jupyter_server.services.config.manager import ConfigManager
@@ -26,8 +27,8 @@ class BeakerIdentityProvider(IdentityProvider):
         config=True
     )
 
-    def _is_authorized_beaker_kernel(self, handler: web.RequestHandler):
-        """Validate Beaker kernel authentication token.
+    def authenticated_beaker_kernel_id(self, handler: web.RequestHandler) -> Optional[str]:
+        """Return the kernel ID from a valid Beaker kernel authentication token.
 
         Checks for a valid Beaker kernel authentication token in the request
         headers and validates it against the kernel's session key using SHA256 hash.
@@ -39,28 +40,34 @@ def _is_authorized_beaker_kernel(self, handler: web.RequestHandler):
 
         Returns
         -------
-        bool
-            True if the token is valid, False otherwise
+        Optional[str]
+            The authenticated kernel ID, or ``None`` when the token is absent
+            or invalid.
         """
         auth_token = handler.request.headers.get(self.beaker_kernel_header, None)
         if not auth_token:
-            return False
+            return None
 
         try:
             preamble, kernel_id, nonce, hash_value = auth_token.split(':')
             if preamble != "beaker-kernel" or not kernel_id or not hash_value:
-                return False
+                return None
             kernel = handler.kernel_manager.get_kernel(kernel_id)
             key = kernel.session.key.decode()
 
             payload = f"{kernel_id}{nonce}{key}".encode()
             reconstructed_hash_value = hashlib.sha256(payload).hexdigest()
-            valid = reconstructed_hash_value == hash_value
-            return valid
+            if not hmac.compare_digest(reconstructed_hash_value, hash_value):
+                return None
+            return kernel_id
 
         except Exception as err:
             logging.error(err)
-            return False
+            return None
+
+    def _is_authorized_beaker_kernel(self, handler: web.RequestHandler):
+        """Validate a Beaker kernel authentication token."""
+        return self.authenticated_beaker_kernel_id(handler) is not None
 
 
     @classmethod
diff --git a/src/beaker_notebook/services/session/__init__.py b/src/beaker_notebook/services/session/__init__.py
index fce5fc42..e88b5968 100644
--- a/src/beaker_notebook/services/session/__init__.py
+++ b/src/beaker_notebook/services/session/__init__.py
@@ -15,6 +15,7 @@ class BeakerSessionManager(SessionManager):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self._sessions: dict[str, dict] = {}
+        self._session_attachment_owners: dict[str, BeakerUser] = {}
 
     @property
     def cursor(self):
@@ -208,6 +209,11 @@ async def save_session(self, session_id, **kwargs):
             a dictionary of the session model
         """
         self._sessions[session_id] = {"session_id": session_id, "started_at": datetime.datetime.now(), **kwargs}
+        kernel_id = kwargs.get("kernel_id")
+        kernel_manager = self.kernel_manager.get_kernel(kernel_id) if kernel_id else None
+        owner = getattr(kernel_manager, "user", None)
+        if isinstance(owner, BeakerUser):
+            self._session_attachment_owners[session_id] = owner
         result = await self.get_session(session_id=session_id)
         return result
 
@@ -240,6 +246,14 @@ async def delete_session(self, session_id=None):
         """Delete a session and shut down its kernel."""
         session = await self.get_session(session_id=session_id)
         kernel_id = session["kernel"]["id"] if session.get("kernel") else None
-        if kernel_id:
-            await ensure_async(self.kernel_manager.shutdown_kernel(kernel_id))
-        self._sessions.pop(session_id, None)
+        kernel_manager = self.kernel_manager.get_kernel(kernel_id) if kernel_id else None
+        owner = getattr(kernel_manager, "user", None) or self._session_attachment_owners.get(session_id)
+        try:
+            if kernel_id:
+                await ensure_async(self.kernel_manager.shutdown_kernel(kernel_id))
+        finally:
+            attachment_manager = getattr(self.parent, "attachment_manager", None)
+            if attachment_manager is not None and session.get("path"):
+                attachment_manager.delete_session(session["path"], owner)
+            self._session_attachment_owners.pop(session_id, None)
+            self._sessions.pop(session_id, None)
diff --git a/tests/test_agent_tools.py b/tests/test_agent_tools.py
index 656844fd..cea2f54d 100644
--- a/tests/test_agent_tools.py
+++ b/tests/test_agent_tools.py
@@ -1,5 +1,7 @@
 """Tests for tools defined on BeakerAgent (lib/agent.py)."""
 
+import json
+
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -166,6 +168,55 @@ async def test_notebook_state_escapes_session_id():
     ET.fromstring(result)  # must still parse
 
 
+# --- session attachments --------------------------------------------------
+
+
+async def test_session_attachments_state_exposes_current_files_and_paths(tmp_path):
+    file_path = tmp_path / "sales.csv"
+    file_path.write_text("region,revenue\nwest,42\n")
+    agent = _agent_instance()
+    agent.context = SimpleNamespace(session_attachments=[{
+        "id": "attachment-one",
+        "name": "sales.csv",
+        "mimetype": "text/csv",
+        "size": file_path.stat().st_size,
+        "kind": "file",
+        "current": True,
+        "path": str(file_path),
+        "root_path": str(tmp_path),
+        "files": ["sales.csv"],
+    }])
+
+    result = json.loads(await BeakerAgent.session_attachments(agent))
+
+    assert result["attachments"][0]["current"] is True
+    assert result["attachments"][0]["path"] == str(file_path)
+    assert "temporary session attachments" in result["instructions"]
+
+
+async def test_get_session_attachment_reads_text_and_blocks_path_escape(tmp_path):
+    root = tmp_path / "archive"
+    root.mkdir()
+    file_path = root / "README.txt"
+    file_path.write_text("temporary attachment")
+    agent = _agent_instance()
+    agent.context = SimpleNamespace(session_attachments=[{
+        "id": "attachment-one",
+        "name": "dataset.zip",
+        "mimetype": "application/zip",
+        "kind": "archive",
+        "root_path": str(root),
+        "original_path": str(tmp_path / "dataset.zip"),
+        "files": ["README.txt"],
+    }])
+
+    result = await BeakerAgent.get_session_attachment(agent, "attachment-one", "README.txt")
+    assert result == "temporary attachment"
+
+    with pytest.raises(ValueError, match="must remain inside"):
+        await BeakerAgent.get_session_attachment(agent, "attachment-one", "../outside.txt")
+
+
 # --- get_notebook_cell ----------------------------------------------------
 
 
diff --git a/tests/test_session_attachments.py b/tests/test_session_attachments.py
new file mode 100644
index 00000000..39b767a5
--- /dev/null
+++ b/tests/test_session_attachments.py
@@ -0,0 +1,292 @@
+import hashlib
+import io
+import zipfile
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from tornado import web
+
+from beaker_notebook.app.api.attachments import SessionAttachmentHandler
+from beaker_notebook.kernel import BeakerKernel
+from beaker_notebook.services.attachments import (
+    AttachmentError,
+    AttachmentLimits,
+    SessionAttachmentManager,
+)
+from beaker_notebook.services.auth import BeakerIdentityProvider
+
+
+@pytest.fixture
+def manager(tmp_path):
+    parent = SimpleNamespace(root_dir=str(tmp_path), virtual_home_root=str(tmp_path))
+    limits = AttachmentLimits(
+        max_upload_bytes=1024 * 1024,
+        max_session_bytes=4 * 1024 * 1024,
+        max_extracted_bytes=2 * 1024 * 1024,
+        max_archive_entries=10,
+        max_archive_ratio=100,
+    )
+    return SessionAttachmentManager(parent, limits=limits)
+
+
+def make_zip(entries: dict[str, bytes]) -> bytes:
+    output = io.BytesIO()
+    with zipfile.ZipFile(output, "w", compression=zipfile.ZIP_DEFLATED) as archive:
+        for name, body in entries.items():
+            archive.writestr(name, body)
+    return output.getvalue()
+
+
+def kernel_auth_handler(kernel_id: str, key: str, token: str):
+    kernel = SimpleNamespace(session=SimpleNamespace(key=key.encode()))
+    kernel_manager = SimpleNamespace(get_kernel=lambda requested: kernel if requested == kernel_id else None)
+    return SimpleNamespace(
+        request=SimpleNamespace(headers={"X-AUTH-BEAKER": token}),
+        kernel_manager=kernel_manager,
+    )
+
+
+def test_kernel_authentication_returns_only_a_verified_kernel_id():
+    provider = BeakerIdentityProvider()
+    kernel_id = "kernel-one"
+    nonce = "1234"
+    key = "secret"
+    digest = hashlib.sha256(f"{kernel_id}{nonce}{key}".encode()).hexdigest()
+    token = f"beaker-kernel:{kernel_id}:{nonce}:{digest}"
+    handler = kernel_auth_handler(kernel_id, key, token)
+
+    assert provider.authenticated_beaker_kernel_id(handler) == kernel_id
+
+    handler.request.headers["X-AUTH-BEAKER"] = f"beaker-kernel:{kernel_id}:{nonce}:forged"
+    assert provider.authenticated_beaker_kernel_id(handler) is None
+
+
+def test_attachment_handler_rejects_an_unverified_kernel_header():
+    identity_provider = SimpleNamespace(
+        beaker_kernel_header="X-AUTH-BEAKER",
+        authenticated_beaker_kernel_id=MagicMock(return_value=None),
+    )
+    handler = SimpleNamespace(
+        request=SimpleNamespace(headers={"X-AUTH-BEAKER": "beaker-kernel:victim:anything:forged"}),
+        identity_provider=identity_provider,
+    )
+
+    with pytest.raises(web.HTTPError, match="Invalid kernel authentication") as error:
+        SessionAttachmentHandler._authenticated_request_kernel_id(handler)
+
+    assert error.value.status_code == 403
+
+
+def test_kernel_clears_attachments_for_its_current_session():
+    kernel = SimpleNamespace(
+        beaker_session="session-one",
+        session_id="fallback-session",
+        jupyter_server="http://localhost:8890",
+        api_auth=MagicMock(return_value="signed-token"),
+    )
+    response = SimpleNamespace(status_code=204, text="")
+
+    with patch("beaker_notebook.kernel.requests.delete", return_value=response) as request:
+        BeakerKernel.clear_session_attachments(kernel)
+
+    request.assert_called_once_with(
+        "http://localhost:8890/beaker/attachments/session-one",
+        headers={"X-AUTH-BEAKER": "signed-token"},
+        timeout=10,
+    )
+
+
+async def test_reset_kernel_clears_session_attachments_before_rebuilding_context():
+    context = SimpleNamespace(
+        SLUG="default",
+        config={"context_info": {}},
+        subkernel=SimpleNamespace(SLUG="python3"),
+        session_attachments=[{"id": "old-attachment"}],
+    )
+    kernel = SimpleNamespace(
+        context=context,
+        clear_session_attachments=MagicMock(),
+        set_context=AsyncMock(),
+        send_set_chat_history=AsyncMock(),
+    )
+    message = SimpleNamespace(header={"msg_id": "reset-one"})
+
+    with patch("beaker_notebook.kernel.reset_config"):
+        await BeakerKernel.reset_kernel.__wrapped__(kernel, message)
+
+    kernel.clear_session_attachments.assert_called_once_with()
+    assert context.session_attachments == []
+    kernel.set_context.assert_awaited_once_with(
+        "default",
+        context.config,
+        subkernel="python3",
+        parent_header=message.header,
+    )
+
+
+def test_regular_attachment_is_draft_until_committed(manager):
+    attachment = manager.create_attachment(
+        "session-one", "sales.csv", "text/csv", b"region,revenue\nwest,42\n"
+    )
+
+    assert attachment["committed"] is False
+    assert attachment["files"] == ["sales.csv"]
+    assert open(attachment["path"], "rb").read() == b"region,revenue\nwest,42\n"
+
+    committed = manager.commit_attachments("session-one", [attachment["id"]])
+    assert [item["id"] for item in committed] == [attachment["id"]]
+    assert committed[0]["committed"] is True
+
+    manager.delete_attachment("session-one", attachment["id"])
+    assert manager.list_attachments("session-one") == []
+
+
+def test_zip_is_extracted_and_original_is_retained(manager):
+    attachment = manager.create_attachment(
+        "session-one",
+        "dataset.zip",
+        "application/zip",
+        make_zip({"data/customers.csv": b"id,name\n1,Ada\n", "README.txt": b"hello"}),
+    )
+
+    assert attachment["archive_status"] == "extracted"
+    assert attachment["file_count"] == 2
+    assert set(attachment["files"]) == {"data/customers.csv", "README.txt"}
+    assert open(attachment["original_path"], "rb").read().startswith(b"PK")
+    assert open(f"{attachment['root_path']}/data/customers.csv", "rb").read() == b"id,name\n1,Ada\n"
+
+
+def test_unsafe_zip_keeps_original_and_reports_extraction_error(manager):
+    attachment = manager.create_attachment(
+        "session-one",
+        "unsafe.zip",
+        "application/zip",
+        make_zip({"../escape.txt": b"nope"}),
+    )
+
+    assert attachment["archive_status"] == "failed"
+    assert "unsafe path" in attachment["archive_error"]
+    assert attachment["files"] == []
+    assert open(attachment["original_path"], "rb").read().startswith(b"PK")
+
+
+def test_upload_and_archive_limits_are_enforced(manager):
+    with pytest.raises(AttachmentError, match="per-file limit"):
+        manager.create_attachment(
+            "session-one", "large.bin", "application/octet-stream", b"x" * (1024 * 1024 + 1)
+        )
+
+    limited = SessionAttachmentManager(
+        manager.parent,
+        limits=AttachmentLimits(
+            max_upload_bytes=1024 * 1024,
+            max_session_bytes=4 * 1024 * 1024,
+            max_extracted_bytes=1024,
+            max_archive_entries=1,
+            max_archive_ratio=100,
+        ),
+    )
+    attachment = limited.create_attachment(
+        "session-two",
+        "many.zip",
+        "application/zip",
+        make_zip({"one.txt": b"one", "two.txt": b"two"}),
+    )
+    assert attachment["archive_status"] == "failed"
+    assert "limit is 1" in attachment["archive_error"]
+
+
+def test_session_cleanup_removes_every_attachment(manager):
+    manager.create_attachment("session-one", "one.txt", "text/plain", b"one")
+    manager.create_attachment("session-one", "two.txt", "text/plain", b"two")
+    assert len(manager.list_attachments("session-one")) == 2
+
+    manager.delete_session("session-one")
+    assert manager.list_attachments("session-one") == []
+
+
+def test_runtime_cleanup_removes_attachments_from_all_sessions(manager):
+    manager.create_attachment("session-one", "one.txt", "text/plain", b"one")
+    manager.create_attachment("session-two", "two.txt", "text/plain", b"two")
+
+    manager.cleanup()
+
+    assert manager.list_attachments("session-one") == []
+    assert manager.list_attachments("session-two") == []
+
+
+def _limits() -> AttachmentLimits:
+    return AttachmentLimits(
+        max_upload_bytes=1024 * 1024,
+        max_session_bytes=4 * 1024 * 1024,
+        max_extracted_bytes=2 * 1024 * 1024,
+        max_archive_entries=10,
+        max_archive_ratio=100,
+    )
+
+
+def test_local_mode_stores_under_shared_data_dir_ignoring_working_dir(tmp_path, monkeypatch):
+    from beaker_notebook.services.auth import BeakerUser
+
+    data_dir = tmp_path / "beaker-data"
+    working_dir = tmp_path / "cwd"
+    monkeypatch.setattr(
+        "beaker_notebook.services.attachments.BEAKER_LOCAL_DATA_PATH", data_dir
+    )
+    parent = SimpleNamespace(
+        root_dir=str(working_dir),
+        virtual_home_root=str(working_dir),
+        local_mode=True,
+    )
+    manager = SessionAttachmentManager(parent, limits=_limits())
+
+    # Even a request carrying a BeakerUser must not scatter files into a home/working dir.
+    user = BeakerUser(username="alice")
+    attachment = manager.create_attachment("session-one", "one.txt", "text/plain", b"one", user)
+
+    stored = Path(attachment["path"])
+    assert data_dir.resolve() / "session-attachments" in stored.resolve().parents
+    assert working_dir not in stored.resolve().parents
+
+
+def test_server_mode_namespaces_attachments_under_the_user_home(tmp_path):
+    from beaker_notebook.services.auth import BeakerUser
+
+    home_root = tmp_path / "homes"
+    parent = SimpleNamespace(
+        root_dir=str(tmp_path / "cwd"),
+        virtual_home_root=str(home_root),
+        local_mode=False,
+    )
+    manager = SessionAttachmentManager(parent, limits=_limits())
+
+    user = BeakerUser(username="alice")
+    attachment = manager.create_attachment("session-one", "one.txt", "text/plain", b"one", user)
+
+    stored = Path(attachment["path"]).resolve()
+    expected_base = (home_root / user.home_dir / ".beaker" / "session-attachments").resolve()
+    assert expected_base in stored.parents
+
+
+def test_local_mode_cleanup_preserves_the_shared_data_dir(tmp_path, monkeypatch):
+    data_dir = tmp_path / "beaker-data"
+    # A sibling artifact the local data dir is expected to also hold.
+    (data_dir / "notebooks").mkdir(parents=True)
+    monkeypatch.setattr(
+        "beaker_notebook.services.attachments.BEAKER_LOCAL_DATA_PATH", data_dir
+    )
+    parent = SimpleNamespace(
+        root_dir=str(tmp_path / "cwd"),
+        virtual_home_root=str(tmp_path / "cwd"),
+        local_mode=True,
+    )
+    manager = SessionAttachmentManager(parent, limits=_limits())
+    manager.create_attachment("session-one", "one.txt", "text/plain", b"one")
+
+    manager.cleanup()
+
+    assert manager.list_attachments("session-one") == []
+    assert data_dir.is_dir()
+    assert (data_dir / "notebooks").is_dir()