Add a new context method (backed by durable workflow attributes) to
propagate your OpenTelemetry context to another workflow:

### PropagateOtelContext

Propagate the current OpenTelemetry context (or optionally, a passed-in
context)
to all workflows started or enqueued in this block so their spans join
the caller's
trace. The propagated context is durably backed by the workflow's
attributes.

Not automatically inherited by child workflows; use PropagateOtelContext
again
inside a workflow to keep its children on the trace.

Usage: 
```py
with PropagateOtelContext():
    handle = queue.enqueue(workflow_function, ...)
```

Closes #798

#801)

Update the workflow outcome write (`update_workflow_outcome`) to apply
only while the workflow's row is still PENDING.

Recording the outcome can result either in a successful recording, the
common path, or, if no rows were affected, a parking of the execution
thread to adopt the final outcome, presumably handled by another
execution thread.

`await_workflow_result` now raises `DBOSNonExistentWorkflowError` when a
result is expected.

Because of the re-ordering of exception raising, we must now
distinguish, in the error catching path, which workflow raised a
cancellation error (like we already do in TS and Java). To this effect,
this PR adds the workflow ID in the cancellation error.

p.s.: This does not solve the race if another execution has resumed and
the workflow is PENDING -- but in this case the outcome should be
deterministic and idempotent.

Adds the following methods providing more flexibility in enqueueing
workflows from the runtime. Closes
#806

```py
def enqueue_workflow_with_options(
    options: EnqueueOptions, *args: Any, **kwargs: Any
) -> WorkflowHandle[Any]:
    """Enqueue a workflow by options, without a reference to its function.

    Takes the same options as :meth:`DBOSClient.enqueue` and builds the same
    row, so the workflow may be implemented by another process, as long as
    it shares this system database. Can safely be called from inside a
    workflow, the enqueued workflow is recorded as a child.

    Unlike :meth:`enqueue_workflow`, options are deliberately not validated
    against the local registry, and ``app_version`` is left unset unless
    given.
    """

async def enqueue_workflow_with_options_async(
    options: EnqueueOptions, *args: Any, **kwargs: Any
) -> WorkflowHandleAsync[Any]:
    """Async version of :meth:`enqueue_workflow_with_options`."""
```

The `DBOSClient` now supports a `lazy` parameter (default `False`). If
`lazy=True`, the system database does not check connection at startup,
but defers the check until the first request is made. A new
`check_connection` method lets you check your connection anytime.

By default, the client retries connection errors. This behavior can now
be configured through the `retry_connection_errors` (default `True`).

This PR adds support for operating multiple applications (potentially in
different languages) in a single system database. All DBOS objects
(workflows, queues, schedules, versions) are now associated with an
application name. If multiple applications share a system database,
their objects are totally isolated by name. By specifying names,
applications sharing a system database can freely interoperate (enqueue
each other's workflows, etc.).

To speed up tests, reset the system database by truncating it instead of
dropping and recreating it.

Closes #812

Also address an issue where bulk fork fails on workflows that have no
steps.

## Description
fork_workflow lacked a timeout option that the TypeScript SDK's
forkWorkflow has (timeoutMS), so forked workflows always ran with no
timeout even when the original workflow had one. This adds a
timeout_seconds parameter across the fork_workflow API surface.

## Implementation
- SystemDatabase.fork_workflow (dbos/_sys_db.py) accepts
workflow_timeout_ms and sets it on the forked row's insert.
- _workflow_commands.fork_workflow accepts timeout_seconds, converts to
ms (validated positive), following the existing seconds-based convention
used by SetWorkflowTimeout.
- DBOS.fork_workflow / fork_workflow_async and DBOSClient.fork_workflow
/ fork_workflow_async pass timeout_seconds through.
- Deadline (workflow_deadline_epoch_ms) is not computed at fork time —
it's computed lazily on dequeue, same as every other queued/enqueued
workflow, so no new deadline logic was needed.
- Matches TS forkWorkflow's timeoutMS behavior; docs for this repo don't
cover fork_workflow params (no docs/ dir here — docs live in the
separate dbos-docs repo), so no doc changes included in this PR.

## Testing
Added test_fork_timeout in tests/test_workflow_management.py:
- Forks a blocked (infinite-loop) workflow with timeout_seconds=0.1,
confirms workflow_timeout_ms is persisted on the forked workflow's
status, and that it actually times out and is cancelled
(DBOSAwaitedWorkflowCancelledError), plus queue cleanup.
- Forks the same workflow with no timeout, confirms workflow_timeout_ms
stays unset.

Ran full tests/test_workflow_management.py locally against SQLite (38
passed) and mypy clean on touched files.

Co-authored-by: Peter Kraft <petereliaskraft@gmail.com>

Batch all database operations during dispatch of dequeued workflows.
Eliminate redundant database writes.

Porting the feature from TS:
https://docs.dbos.dev/typescript/reference/methods#dbosstartworkflow

Queues can now enforce both full-queue and per-partition flow control
limits. New interface:

```python
def register_queue(
    name: str,
    *,
    # Applied to the entire queue
    global_concurrency: Optional[int] = None,
    worker_concurrency: Optional[int] = None,
    limiter: Optional[QueueRateLimit] = None,
    # Applied per-partition
    partition_concurrency: Optional[int] = None,
    partition_worker_concurrency: Optional[int] = None,
    partition_limiter: Optional[QueueRateLimit] = None,
    # Control plane
    polling_interval_sec: float = 1.0,
    on_conflict: QueueConflictResolution = "update_if_latest_version",
) -> Queue:
```

For example, you can create a "fair queue" allows at most one workflow
to run per partition, and at most 10 tasks to run on each worker:

```py
DBOS.register_queue("fair_queue", worker_concurrency=10, partition_concurrency=1)
```

Fully backwards-compatible. Old parameters (`concurrency=`,
`partition_queue=`) are retained, marked deprecated, and mapped to new
parameters.

Closes #821

Closes #818 Closes
#819

Disconnect from Conductor after draining workflows to avoid "zombie"
recovery. Closes #785

- Checkpoint `read_stream`
- Add `read_stream_offset`
- Add timeouts to `read_stream`

You can now specify a timeout for async steps. When the timeout is
reached, the step is cancelled and `DBOSStepTimeoutError` is thrown.

Timeouts are only supported for async steps because Python has no
preemption mechanism for sync steps.

Fix an issue where a duplicate async workflow execution "parking" would
consume a thread.