Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
91 changes: 79 additions & 12 deletions dashscope/applications/finetune/reinforcement/common/model.py
Original file line number Diff line number Diff line change
Expand Up @@ -107,6 +107,10 @@
_LAYER_STATUS_QUERY_SOLUTION = (
"Retry the layer status query after connectivity recovers."
)
_LAYER_STATUS_REJECTED_SOLUTION = (
"Review the server response and layer code. For region lookup errors, "
"report the request ID and layer code to the service owner before retrying."
)
_LAYER_BUILD_FAILED_SOLUTION = (
"Check that requirements.txt or setup.py is valid and every dependency "
"is installable, then review the layer build logs before retrying."
Expand All @@ -121,6 +125,15 @@
)


def _fc_response_rejected(result: Dict[str, Any]) -> bool:
status = result.get("status")
return (
(isinstance(status, dict) and status.get("code", 200) != 200)
or result.get("code") not in (None, "", 0, 200)
or result.get("success") is False
)


class MountStorage(BaseModel):
model_config = ConfigDict(extra="forbid")

Expand Down Expand Up @@ -447,14 +460,27 @@ async def create_layer(
# "signed_url": oss_signed_url,
},
)
if result.get("status", {}).get("code", 500) != 200:
if _fc_response_rejected(result):
raise FunctionLayerError(
f"create rejected by server | {result}",
f"create rejected by server | {deep_mask(result)}",
layer_name=layer_name,
request_id=result.get("request_id"),
body=result,
allow_retry=False,
solution=_LAYER_CREATE_REJECTED_SOLUTION,
)

layer_code = result.get("output", {}).get("layer_code", "")
layer_code = (result.get("output") or {}).get("layer_code", "")
if not layer_code:
raise FunctionLayerError(
f"Empty layer code received: {deep_mask(result)}",
error_code=2013,
layer_name=layer_name,
request_id=result.get("request_id"),
body=result,
allow_retry=False,
solution=_LAYER_CREATE_REJECTED_SOLUTION,
)
logger.debug(
f"Create function layer | layer-name: {layer_name} | "
f"layer_code: {layer_code}",
Expand Down Expand Up @@ -564,7 +590,19 @@ async def get_layer(
solution=_LAYER_STATUS_QUERY_SOLUTION,
) from e

status = result.get("output", {}).get("status", "")
if _fc_response_rejected(result):
raise FunctionLayerError(
f"status query rejected by server | layer_code: "
f"{layer_code} | {deep_mask(result)}",
error_code=2016,
layer_code=layer_code,
request_id=result.get("request_id"),
body=result,
allow_retry=False,
solution=_LAYER_STATUS_REJECTED_SOLUTION,
)

status = (result.get("output") or {}).get("status", "")

if status == "SUCCESS":
logger.debug(
Expand All @@ -579,6 +617,8 @@ async def get_layer(
f"{layer_code} | {result}",
error_code=2015,
layer_code=layer_code,
request_id=result.get("request_id"),
body=result,
allow_retry=False,
solution=_LAYER_BUILD_FAILED_SOLUTION,
)
Expand All @@ -589,6 +629,8 @@ async def get_layer(
f"{layer_code} | {result}",
error_code=2016,
layer_code=layer_code,
request_id=result.get("request_id"),
body=result,
allow_retry=True,
solution=_LAYER_STATUS_MISSING_SOLUTION,
)
Expand Down Expand Up @@ -808,6 +850,8 @@ async def register(
message=f"Function layer deployment failed: {root}",
),
output={},
request_id=e.request_id,
body=e.body,
)
except Exception as e:
root = e
Expand Down Expand Up @@ -934,22 +978,36 @@ async def load(
url,
runtime_dict,
)
self.instance_id = result.get("output", {}).get("instanceId", "")
if _fc_response_rejected(result):
raise FunctionLoadError(
f"load rejected by server | {deep_mask(result)}",
error_code=2023,
entity_id=target_entity_id,
request_id=result.get("request_id"),
body=result,
allow_retry=False,
)

output = result.get("output") or {}
self.instance_id = output.get("instanceId", "")
if not self.instance_id:
raise FunctionLoadError(
f"Empty instance ID received: {result}",
f"Empty instance ID received: {deep_mask(result)}",
error_code=2023,
entity_id=target_entity_id,
request_id=result.get("request_id"),
body=result,
)

self.instance_url = result.get("output", {}).get("trigger_url", "")
self.instance_token = result.get("output", {}).get(
"trigger_token",
"",
)
self.instance_url = output.get("trigger_url", "")
self.instance_token = output.get("trigger_token", "")
if (not self.instance_url) or (not self.instance_token):
raise FunctionLoadError(
"Missing instance URL or token",
error_code=2024,
entity_id=target_entity_id,
request_id=result.get("request_id"),
body=result,
)

logger.info(
Expand Down Expand Up @@ -978,6 +1036,8 @@ async def load(
),
),
output={},
request_id=e.request_id,
body=e.body,
)

except Exception as e:
Expand All @@ -998,6 +1058,8 @@ async def load(
),
),
output={},
request_id=getattr(e, "request_id", None),
body=getattr(e, "body", None),
)

# Perform instance warmup if requested
Expand Down Expand Up @@ -1627,6 +1689,8 @@ async def register_functions(
raise RegistrationError(
reg_result.status.message,
error_code=2052,
request_id=reg_result.request_id,
body=reg_result.body,
)

if fc.type == FunctionType.ROLLOUT:
Expand Down Expand Up @@ -1661,8 +1725,11 @@ async def register_functions(
instance_group_reward_ids.append(instance_id)
else:
raise FunctionLoadError(
f"Load failed: {load_result}",
f"Load failed: {load_result.status.message}",
error_code=2054,
entity_id=entity_id,
request_id=load_result.request_id,
body=load_result.body,
)

except Exception as e:
Expand Down
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
# -*- coding: utf-8 -*-
from enum import Enum
from typing import Optional, List, Any, Dict
from pydantic import BaseModel
from pydantic import BaseModel, Field


class StatusType(str, Enum):
Expand Down Expand Up @@ -91,8 +91,17 @@ class RequestFC(BaseModel):


class ResponseFC(BaseModel):
"""FC result with optional upstream failure context.

``request_id`` and ``body`` retain the original FaaS rejection without
replacing the SDK's existing numeric ``status.code``. The raw ``body``
may contain credentials and is excluded from the model's repr.
"""

status: Status = Status()
output: Optional[Dict[str, Any]] = None
request_id: Optional[str] = None
body: Optional[Dict[str, Any]] = Field(default=None, repr=False)


class RequestTuning(BaseModel):
Expand Down
9 changes: 7 additions & 2 deletions tests/legacy_1x/unit/test_agentic_rl_layer_status.py
Original file line number Diff line number Diff line change
Expand Up @@ -132,8 +132,10 @@ async def test_get_layer_raises_on_region_rejection(monkeypatch):
await FunctionComponentModel().get_layer(LAYER_CODE)

assert "cn-wulanchabu" in str(excinfo.value)
assert excinfo.value.allow_retry is True
assert "Retry the status query" in excinfo.value.solution
assert excinfo.value.allow_retry is False
assert "service owner" in excinfo.value.solution
assert excinfo.value.request_id == LAYER_NOT_FOUND_BODY["request_id"]
assert excinfo.value.body == LAYER_NOT_FOUND_BODY
assert calls["count"] == 1


Expand Down Expand Up @@ -235,6 +237,9 @@ async def test_load_reports_server_rejection_with_context(monkeypatch):
assert ENTITY_ID in result.status.message
assert LAYER_CODE in result.status.message
assert "cn-wulanchabu" in result.status.message
assert "Empty instance ID" not in result.status.message
assert result.request_id == LAYER_NOT_FOUND_BODY["request_id"]
assert result.body == LAYER_NOT_FOUND_BODY


async def test_load_succeeds_once_layer_is_ready(monkeypatch):
Expand Down
8 changes: 6 additions & 2 deletions tests/legacy_1x/unit/test_cli_main.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
from types import SimpleNamespace

import pytest
from typer.main import get_command
from typer.testing import CliRunner

import dashscope
Expand Down Expand Up @@ -111,8 +112,11 @@ def test_rl_register_functions_hyphen_alias_help(self):
)

assert result.exit_code == 0
clean_output = strip_ansi_codes(result.output)
assert "rollout-classpaths" in clean_output
command = get_command(agentic_rl.app).commands["register-functions"]
assert any(
"--rollout-classpaths" in parameter.opts
for parameter in command.params
)

def test_rl_test_functions_hyphen_alias_help(self):
result = CliRunner().invoke(
Expand Down
Loading
Loading