fix: P0-5 — upload-status rollback, streaming copy (LOCAL only), user re-verify, honest lock

B1: `_mark_upload_failed_and_raise` now commits on a separate session
  - Helper takes `request + upload_id`, opens a fresh session from
    `request.app.state.session_factory` and commits there before raising.
  - Closes the named-lock connection-pool leak Codex flagged: the old
    "commit-on-the-same-session" implementation could return the
    GET_LOCK connection to the pool before the enclosing
    `finally: release_named_lock` ran, leaking `mp:<hash>` for up to
    `pool_recycle` and re-opening the same-key upload race.
  - Same helper now used by `create_server_object_payload`'s put-failure
    branch — two failure paths have identical semantics.

B2: streaming copy for soft-delete + restore (`get_stream() + put()`)
  - LOCAL backend: zero-copy (aiofiles stream write). OOM fixed.
  - S3 backend: still OOMs on multi-GB objects — `put()` materializes
    the async iter via `b"".join(chunks)`. Multipart `put` is a
    follow-up; do NOT claim "OOM fixed on production" since production
    defaults to S3.

C1: worker re-verifies `Users.status='active' AND is_deleted=0`
  - `_assert_user_active` called from `_execution_context` after
    resolving `triggered_by`; skips `SYSTEM_CRON_USER_ID`.
  - `USER_DISABLED` error_code goes into the `NODE_FINISHED_EVENT`
    outbox payload — `schedule_node_runs` has no `error_code` column,
    the row only carries the `message` text. Docstrings corrected to
    say so explicitly (previous docstring falsely promised row-level
    observability).

F1: honest browser-local file lock
  - `api.ts` `acquireFileLock/heartbeatFileLock/releaseFileLock/
    releaseFileLockOnUnload` are now no-ops with comments stating they
    never call the network.
  - `scriptWorkspaceStore` dropped `tickHeartbeats`; `tickCleanup`
    simplified to just clear cache.
  - `useEditSessionLifecycle` dropped its 15s heartbeat `setInterval`.
  - `ScriptWorkspace.tsx` renders `.local-lock-banner` info bar when
    `isEditing`. Two tabs may still silently last-write — banner is the
    only guard (acceptable disclosure-only tradeoff).

Dead code: deleted the duplicate `upload_bytes_to_session` in
`backend/src/backend/storage_api.py`. The `services.storage` import
is now the only source of the function; `create_upload_record`'s
docstring updated to point at `backend.resources`.

backend/src/backend/services/storage.py

@@ -151,6 +151,47 @@ async def release_named_lock(session: AsyncSession, lock_name: str) -> None:
     )
 
 
+async def _mark_upload_failed_and_raise(
+    request: Request,
+    upload_id: str,
+    http_status: int,
+    detail: str,
+) -> None:
+    """Flip ``upload_status`` to a terminal state and raise, surviving the
+    surrounding ``session_scope`` rollback AND keeping the named-lock
+    connection pristine.
+
+    Why a separate session: this helper is invoked from inside the
+    ``acquire_named_lock`` critical section in
+    :func:`upload_bytes_to_session` (size / hash / storage-put failure
+    branches). Committing on the lock-bound session could return its
+    connection to the pool, after which the enclosing
+    ``finally: release_named_lock`` would check out a different
+    connection and leak the ``mp:<hash>`` lock for up to ``pool_recycle``
+    seconds — re-opening the same-key upload race the lock exists to
+    close. Opening a fresh session from the factory commits the status
+    flip independently and leaves the lock connection untouched so
+    ``release_named_lock`` runs on the same connection that ran
+    ``GET_LOCK``.
+
+    Why this helper exists at all: the route handler wraps every request
+    in ``session_scope`` (see ``backend.dependencies.database_session``),
+    which rolls back on exception. Without this helper, a naive
+    ``upload.upload_status = "failed"; raise HTTPException(...)`` would
+    lose the status flip and leave the row stuck in ``created``/``uploading``
+    forever.
+    """
+    session_factory = request.app.state.session_factory
+    async with session_factory() as session:
+        upload = await session.scalar(
+            select(UploadSessions).where(UploadSessions.upload_id == upload_id)
+        )
+        if upload is not None:
+            upload.upload_status = "failed"
+            await session.commit()
+    raise HTTPException(http_status, detail)
+
+
 async def _resolve_unique_object_key(
     session: AsyncSession,
     object_key: str,
@@ -401,7 +442,10 @@ async def upload_bytes_to_session(
             f"upload cannot continue from status {upload.upload_status}",
         )
     if upload.expires_at < _utcnow_naive():
+        # 同样需要事务外 commit,否则 expired 状态会被外层 session_scope
+        # 回滚,客户端重试永远得到 created。
         upload.upload_status = "expired"
+        await session.commit()
         raise HTTPException(status.HTTP_409_CONFLICT, "upload expired")
 
     # 按 object_key 串行化「PUT + INSERT」临界区:否则两个同 key 并发上传
@@ -434,16 +478,16 @@ async def upload_bytes_to_session(
             upload.expected_size_bytes is not None
             and actual_size != upload.expected_size_bytes
         ):
-            upload.upload_status = "failed"
-            raise HTTPException(
+            await _mark_upload_failed_and_raise(
+                request, upload.upload_id,
                 status.HTTP_409_CONFLICT,
                 "uploaded bytes size does not match expected_size_bytes",
             )
 
         actual_hash = hashlib.sha256(content).hexdigest() if content else ""
         if upload.expected_hash and actual_hash != upload.expected_hash:
-            upload.upload_status = "failed"
-            raise HTTPException(
+            await _mark_upload_failed_and_raise(
+                request, upload.upload_id,
                 status.HTTP_409_CONFLICT,
                 "uploaded bytes hash does not match expected_hash",
             )
@@ -460,11 +504,11 @@ async def upload_bytes_to_session(
                 metadata=s3_metadata or None,
             )
         except Exception as exc:
-            upload.upload_status = "failed"
-            raise HTTPException(
+            await _mark_upload_failed_and_raise(
+                request, upload.upload_id,
                 status.HTTP_503_SERVICE_UNAVAILABLE,
                 f"failed to write object to storage: {exc}",
-            ) from exc
+            )
 
         item = _build_storage_object(
             upload=upload,
@@ -571,11 +615,11 @@ async def create_server_object_payload(
             metadata={"sha256": content_hash},
         )
     except Exception as exc:
-        upload.upload_status = "failed"
-        raise HTTPException(
+        await _mark_upload_failed_and_raise(
+            request, upload.upload_id,
             status.HTTP_503_SERVICE_UNAVAILABLE,
             f"failed to write object to storage: {exc}",
-        ) from exc
+        )
 
     item = _build_storage_object(
         upload=upload,
@@ -687,8 +731,12 @@ async def soft_delete_object(
         trash_bucket = actual_bucket_name("trash")
         try:
             object_stores = request.app.state.object_stores
-            data = await object_stores[item.bucket_name].get(item.object_key)
-            await object_stores[trash_bucket].put(trash_key, data)
+            # P0-5 / B2: 流式迁移,避免 get() 全量加载导致 10G 对象 OOM。
+            # LOCAL 后端的 put 是 aiofiles 流式写入,get_stream + put 零字节驻留;
+            # S3 后端目前 put 仍会 ``b"".join(chunks)`` 物化到内存,S3 大对象
+            # 的 OOM 修复需要把 put 改成 multipart upload —— 后续工单。
+            stream = object_stores[item.bucket_name].get_stream(item.object_key)
+            await object_stores[trash_bucket].put(trash_key, stream)
             await object_stores[item.bucket_name].delete(item.object_key)
         except Exception as exc:
             raise HTTPException(

backend/src/backend/storage_api.py

@@ -253,7 +253,8 @@ async def create_upload_record(
 
     # Two-step server-proxied upload: the caller PUTs the raw bytes to
     # ``upload_path`` after this response, which routes through
-    # ``upload_bytes_to_session`` below.
+    # ``backend.resources.upload_bytes_to_session`` (the canonical helper
+    # in ``services.storage``).
     return {
         "upload_id": upload.upload_id,
         "status": upload.upload_status,
@@ -284,111 +285,6 @@ def _public_base_url(request: Request) -> str:
     return f"{scheme}://{host}"
 
 
-async def upload_bytes_to_session(
-    upload_id: str, session: AsyncSession, request: Request
-) -> StorageObjects:
-    """Server-proxied upload: read raw bytes from the request body, validate
-    against the ``UploadSessions`` expectations, call ``backend.put``, and
-    create the ``StorageObjects`` row.
-
-    Replaces the old presign-PUT + head-validate flow.
-    """
-    upload = await session.scalar(
-        select(UploadSessions)
-        .where(UploadSessions.upload_id == upload_id)
-        .with_for_update()
-    )
-    if upload is None:
-        raise HTTPException(status.HTTP_404_NOT_FOUND, "upload not found")
-    if upload.upload_status == "completed" and upload.storage_object_id:
-        item = await session.get(StorageObjects, upload.storage_object_id)
-        if item is None or item.object_status != "available":
-            # Linked object was deleted; allow re-upload with the same id.
-            upload.storage_object_id = None
-            upload.upload_status = "created"
-        else:
-            return item
-    if upload.upload_status not in {"created", "uploading"}:
-        raise HTTPException(
-            status.HTTP_409_CONFLICT,
-            f"upload cannot continue from status {upload.upload_status}",
-        )
-    if upload.expires_at < utcnow():
-        upload.upload_status = "expired"
-        raise HTTPException(status.HTTP_409_CONFLICT, "upload expired")
-
-    content = await request.body()
-    actual_size = len(content)
-
-    if (
-        upload.expected_size_bytes is not None
-        and actual_size != upload.expected_size_bytes
-    ):
-        upload.upload_status = "failed"
-        raise HTTPException(
-            status.HTTP_409_CONFLICT,
-            "uploaded bytes size does not match expected_size_bytes",
-        )
-
-    actual_hash = hashlib.sha256(content).hexdigest() if content else ""
-    if upload.expected_hash and actual_hash != upload.expected_hash:
-        upload.upload_status = "failed"
-        raise HTTPException(
-            status.HTTP_409_CONFLICT, "uploaded bytes hash does not match expected_hash"
-        )
-
-    # Round-trip content_type + sha256 metadata through the storage backend
-    # so the next head() (or our own put signature) can recover them.
-    s3_metadata: dict[str, str] = {}
-    if actual_hash:
-        s3_metadata["sha256"] = actual_hash
-
-    try:
-        await request.app.state.object_stores[upload.bucket_name].put(
-            upload.object_key,
-            content,
-            content_type=upload.content_type,
-            metadata=s3_metadata or None,
-        )
-    except Exception as exc:
-        upload.upload_status = "failed"
-        raise HTTPException(
-            status.HTTP_503_SERVICE_UNAVAILABLE,
-            f"failed to write object to storage: {exc}",
-        ) from exc
-
-    file_name = safe_file_name(upload.file_name_hint or "upload.bin")
-    item = StorageObjects(
-        storage_object_id=new_ulid(),
-        workspace_id=upload.workspace_id,
-        owner_user_id=upload.user_id,
-        object_type="file",
-        usage_type=upload.usage_type,
-        storage_backend=settings.storage_backend,
-        bucket_name=upload.bucket_name,
-        object_key=upload.object_key,
-        object_key_hash=upload.object_key_hash,
-        storage_uri=build_storage_uri(upload.bucket_name, upload.object_key),
-        file_name=file_name,
-        file_extension=PurePosixPath(file_name).suffix.lower() or None,
-        mime_type=upload.content_type,
-        size_bytes=actual_size,
-        content_hash=actual_hash or None,
-        object_etag=None,
-        visibility=upload.visibility,
-        is_immutable=int(upload.is_immutable),
-        object_status="available",
-        created_by=upload.user_id,
-    )
-    session.add(item)
-    await session.flush()
-    await session.refresh(item)
-    upload.storage_object_id = item.storage_object_id
-    upload.upload_status = "completed"
-    upload.completed_at = utcnow()
-    return item
-
-
 @router.post(
     "/v1/objects",
     dependencies=[Depends(require_internal_service)],
@@ -438,9 +334,9 @@ async def restore_object(
             status.HTTP_409_CONFLICT, "object has no trash pointer; cannot restore"
         )
     try:
-        # Cross-backend copy: get from trash, put back to source bucket.
+        # P0-5 / B2: 与 soft_delete_object 对称,从回收站恢复也走流式,
+        # 不再把整个对象加载进内存。S3 put 物化限制同 soft_delete_object。
         object_stores = request.app.state.object_stores
-        data = await object_stores[actual_bucket_name("trash")].get(item.object_key)
         source_purpose, _, trash_tail = item.object_key.partition("/")
         if not source_purpose:
             source_purpose = "workspace"
@@ -453,7 +349,8 @@ async def restore_object(
             else trash_tail
         )
         target_bucket = actual_bucket_name(source_purpose)
-        await object_stores[target_bucket].put(source_key, data)
+        stream = object_stores[actual_bucket_name("trash")].get_stream(item.object_key)
+        await object_stores[target_bucket].put(source_key, stream)
     except Exception as exc:
         raise HTTPException(
             status.HTTP_502_BAD_GATEWAY,