Sub-job enqueue cannot set tags, so in-actor fan-out is invisible to tag filters

[rcbevans]

I tag jobs at enqueue time to group everything belonging to one run, so I can answer "what is this run still doing?" and "stop this run" with a single filter. That works for jobs I enqueue from the client, and it works for batch fan-out — but it silently stops working the moment a job is enqueued from inside an actor body.

The three enqueue surfaces disagree:

• JobsClient.enqueue(..., tags=[...]) — has it
• EnqueueItem(..., tags=[...]), so enqueue_batch — has it
• SubJobEnqueuer.enqueue(), i.e. ctx.jobs.enqueue(...) — no tags parameter

There's no inheritance to fall back on either: SubJobEnqueuer is constructed with the loop-scope connection, worker pool, backend, clock and capacity cache, and holds no reference to the enqueuing job, so it has nothing to copy a parent's tags from. Sub-jobs just take the jobs.tags column default of '{}'.

Metadata isn't a workaround for the querying half of this. metadata is accepted on sub-job enqueue, but JobFilter has no metadata predicate — it filters on queue, status, actor, identity_key, batch_id, tags, active — so anything I put in metadata is readable once I already have the row and useless for finding the rows in the first place.

The practical shape of this: a linear pipeline where each stage actor enqueues the next stage with ctx.jobs.enqueue(...), and one stage fans out a batch with ctx.jobs.enqueue_batch(...). The fanned-out jobs are tagged and therefore findable; the stage-to-stage jobs are not, even though both came from the same ctx.jobs object one line apart. So JobFilter(tags=("run:abc",), active=True) returns a systematically incomplete picture of the run, and I end up maintaining a second lookup path — filter by actor name, or by batch_id — purely for the jobs that couldn't be tagged.

This also limits the bulk-cancel primitive discussed in #54: a cancel_where(JobFilter(tags=...)) would not reach in-actor-enqueued jobs, so "cancel this run" stays a two-mechanism operation.

What I'd rather write:

await ctx.jobs.enqueue(
    next_stage,
    payload,
    tags=[f"run:{run_id}", "stage:verify"],
)

While comparing the two signatures, tags isn't the only field that didn't make it across. JobsClient.enqueue also accepts schedule_to_close, start_to_close and heartbeat_timeout, none of which the sub-job path exposes — so a sub-job that needs a different timeout than its actor's declared default can't get one, and has to be enqueued from outside the actor to be configured. (trace_id/span_id I assume are deliberately absent, since context propagates automatically.) The two paths also disagree on the idempotency key type: IdempotencyKey | None on the client, IdempotencyKey | str | None on the sub-job enqueuer.

Open questions from my side:

• Should sub-jobs inherit the enqueuing job's tags by default, with an explicit tags= replacing or merging? Inheritance would fix run-grouping for existing code without any caller change, but it's a behavior change for anyone relying on sub-jobs being untagged, and it means threading the parent job's tags into SubJobEnqueuer.
• Is the absence of schedule_to_close on this path deliberate? A finalizer that snoozes on wait_for_batch for a long time would be killed by one, so "you can't set it from inside an actor" may be a feature rather than an omission — worth documenting either way.
• Is the intended stance that the sub-job surface is deliberately narrower than the client surface, or has it just drifted? If it's deliberate, a note in the docstring about which fields are intentionally unavailable and why would save the comparison.

Happy to be pointed at an existing way to do this if I've missed one.