fix(router): stabilize managed runtime instance identity

[m199369309]

Summary

• stabilize managed Router Runtime instance IDs across Agent reconciliation and Runtime restarts
• derive the Agent-scoped prefix from the assigned listen host while preserving the UUID suffix
• pass the normalized ID through the Runtime environment and every assignment lifecycle status update
• preserve persisted instance IDs when status reports omit the field
• avoid using wildcard listen addresses directly as standalone instance prefixes

Dependency

• Depends on feat(router): add Agent-based Router orchestration #5379 (Router Agent orchestration)

Validation

• pytest -q xinference/router/tests/test_agent.py xinference/router/tests/test_control_plane.py xinference/core/tests/test_router_assignment_store.py
• pre-commit checks for all changed files

deploy/systemd/ is not included.

[gemini-code-assist]

Code Review

This pull request introduces an independent, token-aware router service and agent for DeepSeek-V4 on Xinference, integrating it into the supervisor and adding the necessary REST endpoints, schemas, orchestration, and scheduling components. Feedback on the changes highlights a potential connection leak in restful_api.py when a client disconnects before a stream starts, which can be mitigated using a Starlette BackgroundTask. Additionally, the reviewer suggests preserving the existing pid in router_assignment_store.py when None is passed, introducing a parse_env_float helper in constants.py to safely parse float environment variables, and using explicit is None checks in router_node_store.py to avoid overwriting empty collections.

[gemini-code-assist]

If the client disconnects before the stream starts (i.e., before Starlette calls body_stream), body_stream is never executed, and upstream_response.aclose() is never called, leading to a leaked connection. Use a Starlette BackgroundTask to ensure the connection is closed even if the stream never starts.

Suggested change

	return StreamingResponse(
	body_stream(),
	status_code=upstream_response.status_code,
	headers=response_headers,
	)
	return StreamingResponse(
	body_stream(),
	status_code=upstream_response.status_code,
	headers=response_headers,
	background=BackgroundTask(upstream_response.aclose),
	)

References

1. When managing critical concurrency resources (such as capacity gates or runtime snapshots) in a streaming response, use a Starlette BackgroundTask as an early-disconnect fallback alongside generator finally blocks. Ensure the cleanup task is idempotent, shielded from cancellation, and guarantees release of all resources even if the client disconnects before the stream starts.

[gemini-code-assist]

In report_status, if pid is not provided (defaults to None), it will overwrite the existing pid in the database with NULL. We should preserve the existing pid if None is passed, similar to how instance_id is handled.

Suggested change

	effective_instance_id = (
	current["instance_id"] if instance_id is None else instance_id
	)
	now = self._now()
	with self._lock, self._connect() as conn:
	conn.execute(
	"""UPDATE token_router_assignments
	SET observed_state = ?, pid = ?, instance_id = ?, last_error = ?,
	observed_json = ?, last_seen_at = ?, updated_at = ?
	WHERE assignment_id = ?""",
	(
	observed_state,
	pid,
	effective_instance_id,
	effective_instance_id = (
	current["instance_id"] if instance_id is None else instance_id
	)
	effective_pid = current["pid"] if pid is None else pid
	now = self._now()
	with self._lock, self._connect() as conn:
	conn.execute(
	"""UPDATE token_router_assignments
	SET observed_state = ?, pid = ?, instance_id = ?, last_error = ?,
	observed_json = ?, last_seen_at = ?, updated_at = ?
	WHERE assignment_id = ?""",
	(
	observed_state,
	effective_pid,
	effective_instance_id,

[gemini-code-assist]

Import BackgroundTask from starlette.background to support registering a background cleanup task for the streaming response.

Suggested change

	from starlette.responses import PlainTextResponse, StreamingResponse
	from starlette.background import BackgroundTask
	from starlette.responses import PlainTextResponse, StreamingResponse

[gemini-code-assist]

Add a parse_env_float helper function to safely parse float environment variables with a fallback default, logging a warning on misconfiguration instead of crashing on startup.

def parse_env_bool(name: str, default: bool) -> bool: 
    """Parse a boolean environment variable without accepting ambiguous values."""
    raw = os.environ.get(name)
    if raw is None or not raw.strip():
        return default
    normalized = raw.strip().lower()
    if normalized in {"1", "true", "yes", "on"}:
        return True
    if normalized in {"0", "false", "no", "off"}:
        return False
    raise ValueError(
        f"{name} must be one of true/false, 1/0, yes/no, or on/off; got {raw!r}"
    )


def parse_env_float(name: str, default: float) -> float:
    """Safely parse a float environment variable with a fallback default."""
    raw = os.environ.get(name)
    if raw is None or not raw.strip():
        return default
    try:
        return float(raw)
    except ValueError:
        import logging
        logging.getLogger("xinference").warning(
            f"Environment variable {name} is misconfigured (got {raw!r}). "
            f"Falling back to default: {default}"
        )
        return default

References

1. When safely parsing environment variables with a fallback default to prevent startup crashes, log a warning so that operators are notified of the misconfiguration instead of silently falling back.

[gemini-code-assist]

Use the newly introduced parse_env_float helper to safely parse float environment variables with a fallback default, preventing startup crashes on misconfiguration.

Suggested change

	XINFERENCE_TOKEN_ROUTER_HEARTBEAT_TIMEOUT_SECONDS = float(
	os.environ.get("XINFERENCE_TOKEN_ROUTER_HEARTBEAT_TIMEOUT_SECONDS", "90")
	)
	XINFERENCE_TOKEN_ROUTER_AGENT_SUSPECT_SECONDS = float(
	os.environ.get("XINFERENCE_TOKEN_ROUTER_AGENT_SUSPECT_SECONDS", "30")
	)
	XINFERENCE_TOKEN_ROUTER_AGENT_OFFLINE_SECONDS = float(
	os.environ.get("XINFERENCE_TOKEN_ROUTER_AGENT_OFFLINE_SECONDS", "45")
	)
	XINFERENCE_TOKEN_ROUTER_AGENT_MONITOR_SECONDS = float(
	os.environ.get("XINFERENCE_TOKEN_ROUTER_AGENT_MONITOR_SECONDS", "5")
	)
	XINFERENCE_TOKEN_ROUTER_STALE_RETENTION_SECONDS = float(
	os.environ.get("XINFERENCE_TOKEN_ROUTER_STALE_RETENTION_SECONDS", "300")
	)
	XINFERENCE_TOKEN_ROUTER_HEARTBEAT_TIMEOUT_SECONDS = parse_env_float(
	"XINFERENCE_TOKEN_ROUTER_HEARTBEAT_TIMEOUT_SECONDS", 90.0
	)
	XINFERENCE_TOKEN_ROUTER_AGENT_SUSPECT_SECONDS = parse_env_float(
	"XINFERENCE_TOKEN_ROUTER_AGENT_SUSPECT_SECONDS", 30.0
	)
	XINFERENCE_TOKEN_ROUTER_AGENT_OFFLINE_SECONDS = parse_env_float(
	"XINFERENCE_TOKEN_ROUTER_AGENT_OFFLINE_SECONDS", 45.0
	)
	XINFERENCE_TOKEN_ROUTER_AGENT_MONITOR_SECONDS = parse_env_float(
	"XINFERENCE_TOKEN_ROUTER_AGENT_MONITOR_SECONDS", 5.0
	)
	XINFERENCE_TOKEN_ROUTER_STALE_RETENTION_SECONDS = parse_env_float(
	"XINFERENCE_TOKEN_ROUTER_STALE_RETENTION_SECONDS", 300.0
	)

References

1. When safely parsing environment variables with a fallback default to prevent startup crashes, log a warning so that operators are notified of the misconfiguration instead of silently falling back.

[gemini-code-assist]

Using or for fallback (data.get("reported_labels") or data.get("labels", {})) will cause reported_labels to silently fall back to labels if an empty dictionary {} is explicitly provided. Use an explicit is None check instead to preserve empty collections.

        reported_labels_val = data.get("reported_labels")
        if reported_labels_val is None:
            reported_labels_val = data.get("labels", {})
        reported_labels = dict(reported_labels_val)

References

1. Use explicit is None checks instead of truthiness checks (e.g., if not config) when validating configuration payloads where empty collections (like [] or {}) are meaningful inputs that should be preserved and handled/validated downstream, rather than silently falling back to default behaviors.

[m199369309]

Addressed the review feedback in commit 33598105:

• Added a BackgroundTask cleanup fallback for streamed upstream responses while retaining generator finally cleanup.
• Preserved an existing assignment PID when a status report omits pid.
• Added finite, warning-backed parsing for Token Router float environment settings.
• Preserved the distinction between omitted and explicitly empty reported_labels at the API and store layers.
• Added regression coverage for all four behaviors.

Validation: targeted Token Router tests pass (70 passed); pre-commit hooks pass.