[Bug] `SandboxImportNotificationPolicy.WARN_ON_UNINTENTIONAL_PASSTHROUGH` warns on import of the workflow itself to the sandbox

Inizia da temporalio/worker/workflow_sandbox/_importer.py, in prossimità dell’avviso alla riga 323, quindi esegui la riproduzione minima del worker riportata nell’issue con WARN_ON_UNINTENTIONAL_PASSTHROUGH abilitato. Traccia il modo in cui il modulo del workflow viene caricato nella sandbox e assicurati che il caricamento del workflow stesso non produca l’avviso di unintentional-passthrough, mantenendo invariato il comportamento di avviso richiesto.

What are you really trying to do?

Leverage SandboxImportNotificationPolicy.WARN_ON_UNINTENTIONAL_PASSTHROUGH to identify imports that aren't intentionally passed through to the sandbox.

Describe the bug

Using a sandboxed runner with the workflows defined in a separate from the worker triggers the unintentional passthrough warning. The workflow must be loaded into the sandbox each time and therefore should be exempt from this warning.

Minimal Reproduction

# workflows.py
from datetime import timedelta

from temporalio import workflow

from warning.activities import echo_activity


@workflow.defn
class ExampleWorkflow:
    @workflow.run
    async def run(self) -> str:
        activity_result = await workflow.execute_activity(
            echo_activity,
            "example",
            task_queue="example-task-queue",
            start_to_close_timeout=timedelta(seconds=10),
        )

        return activity_result

# activities.py
from temporalio import activity


@activity.defn
async def echo_activity(input: str) -> str:
    return f"hello {input}"

# worker.py
import asyncio

from temporalio.client import Client
from temporalio.envconfig import ClientConfig
from temporalio.worker import Worker
from temporalio.worker.workflow_sandbox import (
    SandboxedWorkflowRunner,
    SandboxRestrictions,
)
from temporalio.workflow import SandboxImportNotificationPolicy

from warning.activities import echo_activity
from warning.workflows import ExampleWorkflow


async def main():
    config = ClientConfig.load_client_connect_config()
    config.setdefault("target_host", "localhost:7233")

    # Start client
    client = await Client.connect(**config)

    worker = Worker(
        client,
        task_queue="example-task-queue",
        workflows=[ExampleWorkflow],
        activities=[echo_activity],
        workflow_runner=SandboxedWorkflowRunner(
            restrictions=SandboxRestrictions.default.with_import_notification_policy(
                SandboxImportNotificationPolicy.WARN_ON_DYNAMIC_IMPORT
                | SandboxImportNotificationPolicy.WARN_ON_UNINTENTIONAL_PASSTHROUGH
            )
        ),
    )

    await worker.run()


if __name__ == "__main__":
    asyncio.run(main())

Starting the above worker yields this log on startup and workflow execution:

uv run warning/worker.py
/.../temporalio/worker/workflow_sandbox/_importer.py:323: UserWarning: Module warning.workflows was not intentionally passed through to the sandbox.
warnings.warn(
/.../temporalio/worker/workflow_sandbox/_importer.py:323: UserWarning: Module warning.activities was not intentionally passed through to the sandbox.
warnings.warn(

Environment/Versions

• Temporal Version: SDK 1.20

Additional context