diff --git a/.github/workflows/am-sdk-publish.yml b/.github/workflows/am-sdk-publish.yml index 83c0b2fc3..ecb094b38 100644 --- a/.github/workflows/am-sdk-publish.yml +++ b/.github/workflows/am-sdk-publish.yml @@ -172,12 +172,22 @@ jobs: github - ${env.GITHUB_ACTOR} + x-access-token ${env.GITHUB_TOKEN} github-core - ${env.GITHUB_ACTOR} + x-access-token + ${env.GITHUB_TOKEN} + + + github-investment + x-access-token + ${env.GITHUB_TOKEN} + + + github-trade + x-access-token ${env.GITHUB_TOKEN} @@ -194,8 +204,7 @@ jobs: PL=$(IFS=, ; echo "${MODULES[*]}") mvn -B clean verify -pl "$PL" -am -DskipTests env: - GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} - GITHUB_ACTOR: ${{ github.actor }} + GITHUB_TOKEN: ${{ secrets.GHCR_TOKEN || secrets.GITHUB_PACKAGES_TOKEN || secrets.GITHUB_TOKEN }} publish-analysis-client-sdk: name: Publish am-analysis-client (java-sdk) diff --git a/.github/workflows/core-libraries-publish.yml b/.github/workflows/core-libraries-publish.yml index 8759f6e89..c2be129b1 100644 --- a/.github/workflows/core-libraries-publish.yml +++ b/.github/workflows/core-libraries-publish.yml @@ -197,22 +197,22 @@ jobs: github - ${env.GITHUB_ACTOR} + x-access-token ${env.GITHUB_TOKEN} github-core - ${env.GITHUB_ACTOR} + x-access-token ${env.GITHUB_TOKEN} github-investment - ${env.GITHUB_ACTOR} + x-access-token ${env.GITHUB_TOKEN} github-trade - ${env.GITHUB_ACTOR} + x-access-token ${env.GITHUB_TOKEN} @@ -221,8 +221,7 @@ jobs: - name: Reactor verify (-pl -am) run: mvn -B clean verify -pl "${{ needs.detect-changes.outputs.maven-pl-list }}" -am -DskipTests env: - GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} - GITHUB_ACTOR: ${{ github.actor }} + GITHUB_TOKEN: ${{ secrets.GHCR_TOKEN || secrets.GITHUB_PACKAGES_TOKEN || secrets.GITHUB_TOKEN }} # ── main/master: publish parent POM (non-recursive) ───────────────────── publish-root-pom: @@ -255,22 +254,22 @@ jobs: github - ${env.GITHUB_ACTOR} + x-access-token ${env.GITHUB_TOKEN} github-core - ${env.GITHUB_ACTOR} + x-access-token ${env.GITHUB_TOKEN} github-investment - ${env.GITHUB_ACTOR} + x-access-token ${env.GITHUB_TOKEN} github-trade - ${env.GITHUB_ACTOR} + x-access-token ${env.GITHUB_TOKEN} @@ -297,8 +296,7 @@ jobs: } cat deploy.log env: - GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} - GITHUB_ACTOR: ${{ github.actor }} + GITHUB_TOKEN: ${{ secrets.GHCR_TOKEN || secrets.GITHUB_PACKAGES_TOKEN || secrets.GITHUB_TOKEN }} - name: Advance parent to next snapshot and push if: github.ref == 'refs/heads/main' || github.ref == 'refs/heads/master' run: | diff --git a/.github/workflows/qa-agent-notify.yml b/.github/workflows/qa-agent-notify.yml index 7dd579974..06af0fab7 100644 --- a/.github/workflows/qa-agent-notify.yml +++ b/.github/workflows/qa-agent-notify.yml @@ -127,9 +127,12 @@ jobs: needs: [detect, wait-catalog] if: always() && needs.detect.result == 'success' && needs.detect.outputs.skip != 'true' && needs.wait-catalog.result == 'success' runs-on: [self-hosted, linux, x64, docker] + outputs: + tracking_id: ${{ steps.trigger.outputs.tracking_id }} steps: - uses: actions/checkout@v4 - name: Trigger qa-agent + id: trigger env: QA_AGENT_BASE_URL: http://am-qa-agents.am-apps-dev.svc.cluster.local:8150 QA_AGENT_GATEWAY_TOKEN: ${{ secrets.QA_AGENT_GATEWAY_TOKEN }} @@ -145,3 +148,21 @@ jobs: exit 1 fi python3 scripts/qa_agent_notify_ci.py notify + + wait-verify: + name: Wait verify (SPT/UI) + needs: [detect, notify] + if: always() && needs.detect.result == 'success' && needs.detect.outputs.skip != 'true' && needs.notify.result == 'success' && needs.notify.outputs.tracking_id != '' + runs-on: [self-hosted, linux, x64, docker] + steps: + - uses: actions/checkout@v4 + - name: Poll until releasable decision + env: + QA_AGENT_BASE_URL: http://am-qa-agents.am-apps-dev.svc.cluster.local:8150 + QA_AGENT_GATEWAY_TOKEN: ${{ secrets.QA_AGENT_GATEWAY_TOKEN }} + TRACKING_ID: ${{ needs.notify.outputs.tracking_id }} + QA_AGENT_VERIFY_TIMEOUT_SEC: "1200" + QA_AGENT_VERIFY_POLL_SEC: "10" + run: | + set -euo pipefail + python3 scripts/qa_agent_notify_ci.py wait-verify diff --git a/libraries/am-trade-client-lib/pom.xml b/libraries/am-trade-client-lib/pom.xml index 41787a013..d78c078ca 100644 --- a/libraries/am-trade-client-lib/pom.xml +++ b/libraries/am-trade-client-lib/pom.xml @@ -1,4 +1,5 @@ + 1.0.0-SNAPSHOT - + com.am.domain am-trade-domain - ${project.version} + 1.1.3-SNAPSHOT @@ -56,4 +57,3 @@ - diff --git a/libraries/am-trade-client-lib/src/main/java/com/am/trade/client/auth/TradeAuthTokenSupplier.java b/libraries/am-trade-client-lib/src/main/java/com/am/trade/client/auth/TradeAuthTokenSupplier.java new file mode 100644 index 000000000..99adb7229 --- /dev/null +++ b/libraries/am-trade-client-lib/src/main/java/com/am/trade/client/auth/TradeAuthTokenSupplier.java @@ -0,0 +1,12 @@ +package com.am.trade.client.auth; + +/** + * Optional per-request bearer for outbound trade HTTP calls. + * Host apps (e.g. am-mcp-server {@code AuthTokenProvider}) supply the live user JWT + * so each TradeClientService request can authenticate without a static apiKey. + */ +@FunctionalInterface +public interface TradeAuthTokenSupplier { + /** @return bearer token for the current request, or null/blank if unavailable */ + String getToken(); +} diff --git a/libraries/am-trade-client-lib/src/main/java/com/am/trade/client/service/TradeClientService.java b/libraries/am-trade-client-lib/src/main/java/com/am/trade/client/service/TradeClientService.java index 72652fb1a..36da872c3 100644 --- a/libraries/am-trade-client-lib/src/main/java/com/am/trade/client/service/TradeClientService.java +++ b/libraries/am-trade-client-lib/src/main/java/com/am/trade/client/service/TradeClientService.java @@ -3,8 +3,9 @@ import am.trade.sdk.AmTradeSdk; import com.am.domain.trade.TradePortfolio; import com.am.domain.trade.TradeTransaction; -import lombok.RequiredArgsConstructor; +import com.am.trade.client.auth.TradeAuthTokenSupplier; import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.ObjectProvider; import org.springframework.stereotype.Service; import java.math.BigDecimal; @@ -14,22 +15,30 @@ import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.function.Supplier; @Service @Slf4j -@RequiredArgsConstructor public class TradeClientService { private final AmTradeSdk tradeSdk; + private final ObjectProvider tokenSupplier; + + public TradeClientService(AmTradeSdk tradeSdk, ObjectProvider tokenSupplier) { + this.tradeSdk = tradeSdk; + this.tokenSupplier = tokenSupplier; + } public List getPortfolios(String userId) { - try { - Map response = tradeSdk.getPortfolioClient().getAllPortfolios(0, 100); - return mapPortfolios(response); - } catch (Exception e) { - log.error("Failed to fetch trade portfolios for user: {}", userId, e); - return Collections.emptyList(); - } + return withAuth(() -> { + try { + Map response = tradeSdk.getPortfolioClient().getAllPortfolios(0, 100); + return mapPortfolios(response); + } catch (Exception e) { + log.error("Failed to fetch trade portfolios for user: {}", userId, e); + return Collections.emptyList(); + } + }); } public List getRecentTrades(String userId) { @@ -37,50 +46,78 @@ public List getRecentTrades(String userId) { } public List getRecentTrades(String userId, int page, int size) { - try { - Map response = tradeSdk.getTradeClient().getTradesByFreeTab(page, size); - return mapTrades(response, userId); - } catch (Exception e) { - log.error("Failed to fetch recent trades for user: {}", userId, e); - return Collections.emptyList(); - } + return withAuth(() -> { + try { + Map response = tradeSdk.getTradeClient().getTradesByFreeTab(page, size); + return mapTrades(response, userId); + } catch (Exception e) { + log.error("Failed to fetch recent trades for user: {}", userId, e); + return Collections.emptyList(); + } + }); } public List getTradesBySymbol(String userId, String symbol, int page, int size) { - try { - Map response = tradeSdk.getTradeClient() - .getTradesByFreeTabAndSymbol(symbol, page, size); - return mapTrades(response, userId); - } catch (Exception e) { - log.error("Failed to fetch trades for symbol {}: {}", symbol, e.getMessage()); - return Collections.emptyList(); - } + return withAuth(() -> { + try { + Map response = tradeSdk.getTradeClient() + .getTradesByFreeTabAndSymbol(symbol, page, size); + return mapTrades(response, userId); + } catch (Exception e) { + log.error("Failed to fetch trades for symbol {}: {}", symbol, e.getMessage()); + return Collections.emptyList(); + } + }); } public Map filterTrades(Map filters) { - try { - return tradeSdk.getTradeClient().filterTrades(filters); - } catch (Exception e) { - log.error("Failed to filter trades: {}", e.getMessage()); - return Map.of("error", e.getMessage() != null ? e.getMessage() : "filter failed"); - } + return withAuth(() -> { + try { + return tradeSdk.getTradeClient().filterTrades(filters); + } catch (Exception e) { + log.error("Failed to filter trades: {}", e.getMessage()); + return Map.of("error", e.getMessage() != null ? e.getMessage() : "filter failed"); + } + }); } public Map getTradeMetrics(String portfolioId) { - try { - return tradeSdk.getAnalyticsClient().getTradeMetrics(portfolioId); - } catch (Exception e) { - log.error("Failed to fetch trade metrics for {}: {}", portfolioId, e.getMessage()); - return Map.of("error", e.getMessage() != null ? e.getMessage() : "metrics failed"); - } + return withAuth(() -> { + try { + return tradeSdk.getAnalyticsClient().getTradeMetrics(portfolioId); + } catch (Exception e) { + log.error("Failed to fetch trade metrics for {}: {}", portfolioId, e.getMessage()); + return Map.of("error", e.getMessage() != null ? e.getMessage() : "metrics failed"); + } + }); } public Map getTradePortfolioSummary(String portfolioId) { - try { - return tradeSdk.getPortfolioClient().getPortfolioSummary(portfolioId); - } catch (Exception e) { - log.error("Failed to fetch trade portfolio summary for {}: {}", portfolioId, e.getMessage()); - return Map.of("error", e.getMessage() != null ? e.getMessage() : "summary failed"); + return withAuth(() -> { + try { + return tradeSdk.getPortfolioClient().getPortfolioSummary(portfolioId); + } catch (Exception e) { + log.error("Failed to fetch trade portfolio summary for {}: {}", portfolioId, e.getMessage()); + return Map.of("error", e.getMessage() != null ? e.getMessage() : "summary failed"); + } + }); + } + + private T withAuth(Supplier action) { + TradeAuthTokenSupplier supplier = tokenSupplier.getIfAvailable(); + if (supplier == null) { + return action.get(); + } + Object lock = tradeSdk.getConfiguration(); + synchronized (lock) { + String previous = tradeSdk.getConfiguration().getApiKey(); + try { + String token = supplier.getToken(); + tradeSdk.getConfiguration().setApiKey(token != null ? token : ""); + return action.get(); + } finally { + tradeSdk.getConfiguration().setApiKey(previous); + } } } diff --git a/pom.xml b/pom.xml index 751d31927..64e6df68e 100644 --- a/pom.xml +++ b/pom.xml @@ -229,16 +229,16 @@ - - github-investment - GitHub Packages - Core Services - https://maven.pkg.github.com/AM-Portfolio/am-core-services - github-trade GitHub Packages - Trade Services https://maven.pkg.github.com/AM-Portfolio/am-trade-services + + github-investment + GitHub Packages - Core Services + https://maven.pkg.github.com/AM-Portfolio/am-core-services + diff --git a/scripts/qa_agent_notify_ci.py b/scripts/qa_agent_notify_ci.py index bead69c9f..40a8e309c 100644 --- a/scripts/qa_agent_notify_ci.py +++ b/scripts/qa_agent_notify_ci.py @@ -59,8 +59,10 @@ def enabled(svc: str): return False, "missing_spt_yaml", str(data.get("environment") or "dev"), True return True, "ok", str(data.get("environment") or "dev"), req - service = input_svc - if not service and os.environ.get("GITHUB_EVENT_NAME") == "push": + matched: list[str] = [] + if input_svc: + matched = [input_svc] + elif os.environ.get("GITHUB_EVENT_NAME") == "push": try: diff = subprocess.check_output( ["git", "diff", "--name-only", "HEAD~1", "HEAD"], text=True @@ -69,16 +71,18 @@ def enabled(svc: str): diff = "" for line in diff.splitlines(): for c in candidates: - if line.startswith(f"services/{c}/"): - service = c - break - if service: - break - if not service: - service = "am-analysis" + if line.startswith(f"services/{c}/") and c not in matched: + matched.append(c) + if not matched: + matched = ["am-analysis"] + # Primary = first matched (workflow jobs use single service today) + service = matched[0] ok, reason, env, req = enabled(service) + # Also emit all enabled matched services for future fan-out + enabled_services = [s for s in matched if enabled(s)[0]] print(f"service={service}") + print(f"services={','.join(enabled_services) if enabled_services else service}") print(f"environment={env}") print(f"require_catalog={'true' if req else 'false'}") print("skip=false" if ok else "skip=true") @@ -86,6 +90,22 @@ def enabled(svc: str): return 0 +def _http_json(url: str, *, method: str = "GET", data: dict | None = None, token: str | None = None, timeout: int = 30): + headers = {"Accept": "application/json"} + body = None + if data is not None: + body = json.dumps(data).encode("utf-8") + headers["Content-Type"] = "application/json" + if token: + headers["Authorization"] = f"Bearer {token}" + req = urllib.request.Request(url, data=body, method=method, headers=headers) + try: + with urllib.request.urlopen(req, timeout=timeout) as resp: + return resp.status, resp.read().decode("utf-8", errors="replace") + except urllib.error.HTTPError as e: + return e.code, e.read().decode("utf-8", errors="replace") + + def cmd_wait() -> int: base = os.environ["QA_AGENT_BASE_URL"].rstrip("/") service = os.environ["SERVICE"] @@ -93,13 +113,7 @@ def cmd_wait() -> int: print(f"Waiting for {service} in {url}") for i in range(1, 37): try: - req = urllib.request.Request(url, method="GET") - with urllib.request.urlopen(req, timeout=10) as resp: - code = resp.status - body = resp.read().decode("utf-8", errors="replace") - except urllib.error.HTTPError as e: - code = e.code - body = e.read().decode("utf-8", errors="replace") + code, body = _http_json(url, timeout=10) except Exception as e: print(f"attempt {i}: error {type(e).__name__}: {e}") time.sleep(5) @@ -127,6 +141,17 @@ def cmd_wait() -> int: def cmd_notify() -> int: base = os.environ["QA_AGENT_BASE_URL"].rstrip("/") token = os.environ["QA_AGENT_GATEWAY_TOKEN"] + # Fail fast if Specs cannot run k6 (release SPT would soft-fail / miss evidence) + try: + hc, hb = _http_json(f"{base}/api/platform/health", timeout=10) + if hc == 200: + health = json.loads(hb) + if health.get("k6_binary") is False: + print("::error::qa-agent k6_binary=false — rebuild image with k6 before notify", file=sys.stderr) + return 1 + except Exception as e: + print(f"::warning::platform health check failed: {e}") + payload = { "repo": os.environ["REPO"], "branch": os.environ["BRANCH"], @@ -140,22 +165,11 @@ def cmd_notify() -> int: } url = f"{base}/v2/workflows/release-readiness" print(f"POST {url}") - req = urllib.request.Request( - url, - data=json.dumps(payload).encode("utf-8"), - method="POST", - headers={ - "Authorization": f"Bearer {token}", - "Content-Type": "application/json", - }, - ) try: - with urllib.request.urlopen(req, timeout=60) as resp: - code = resp.status - body = resp.read().decode("utf-8", errors="replace") - except urllib.error.HTTPError as e: - code = e.code - body = e.read().decode("utf-8", errors="replace") + code, body = _http_json(url, method="POST", data=payload, token=token, timeout=60) + except Exception as e: + print(f"::error::qa-agent notify request failed: {e}", file=sys.stderr) + return 1 print(f"HTTP {code}") print(body) if code < 200 or code >= 300: @@ -168,13 +182,106 @@ def cmd_notify() -> int: if data.get("skipped") is True: print(f"::warning::qa-agent skipped: {data.get('reason')}") return 0 + tid = data.get("tracking_id") or "" + print(f"tracking_id={tid}") + gh_out = os.environ.get("GITHUB_OUTPUT") + if gh_out and tid: + with open(gh_out, "a", encoding="utf-8") as f: + f.write(f"tracking_id={tid}\n") print(f"qa-agent activated for {os.environ['SERVICE']} on {os.environ['BRANCH']}") return 0 +def cmd_wait_verify() -> int: + """Poll until Temporal verify finishes (awaiting_release step), then fail if not releasable. + + Does not wait for HITL approve/reject. + """ + base = os.environ["QA_AGENT_BASE_URL"].rstrip("/") + token = os.environ.get("QA_AGENT_GATEWAY_TOKEN") or "" + tid = (os.environ.get("TRACKING_ID") or "").strip() + if not tid: + print("::error::TRACKING_ID required for wait-verify", file=sys.stderr) + return 1 + timeout_sec = int(os.environ.get("QA_AGENT_VERIFY_TIMEOUT_SEC") or "1200") + poll = int(os.environ.get("QA_AGENT_VERIFY_POLL_SEC") or "10") + url = f"{base}/v2/runs/{tid}" + print(f"Waiting for verify on {url} (timeout={timeout_sec}s)") + deadline = time.time() + timeout_sec + attempt = 0 + while time.time() < deadline: + attempt += 1 + try: + code, body = _http_json(url, token=token or None, timeout=20) + except Exception as e: + print(f"attempt {attempt}: error {type(e).__name__}: {e}") + time.sleep(poll) + continue + if code == 404: + print(f"attempt {attempt}: run not found yet") + time.sleep(poll) + continue + if code >= 400: + print(f"attempt {attempt}: http={code} body={body[:500]}") + time.sleep(poll) + continue + try: + data = json.loads(body) + except json.JSONDecodeError: + print(f"attempt {attempt}: bad json") + time.sleep(poll) + continue + steps = data.get("steps") or {} + status = str(data.get("status") or "") + awaiting = steps.get("awaiting_release") + post_verify = steps.get("post_test_verify") or steps.get("verify") + # Prefer post_test_verify (available before PDF/HITL); then awaiting_release + releasable = None + blockers = None + if isinstance(post_verify, dict) and "releasable" in post_verify: + releasable = post_verify.get("releasable") + blockers = post_verify.get("blockers") + elif isinstance(awaiting, dict) and "releasable" in awaiting: + releasable = awaiting.get("releasable") + blockers = awaiting.get("blockers") + terminal_fail = status in { + "failed", + "error", + "release_rejected", + "hitl_timeout", + } + if post_verify is not None or awaiting is not None or releasable is not None or terminal_fail: + print(f"verify reached status={status} releasable={releasable} blockers={blockers}") + print(body[:4000]) + if terminal_fail and releasable is not True: + print("::error::release-readiness terminal failure", file=sys.stderr) + return 1 + if releasable is False: + print("::error::release-readiness not releasable (SPT/UI/verify blockers)", file=sys.stderr) + return 1 + if releasable is True: + print("verify passed (releasable=true); HITL still pending if awaiting_release") + return 0 + # awaiting without releasable key — treat as incomplete + if awaiting is not None and releasable is None: + print("::warning::awaiting_release present but releasable missing; checking matrix steps") + # Also inspect execute_matrix / post_test_verify nested + exec_step = steps.get("execute_matrix") or {} + if isinstance(exec_step, dict) and exec_step.get("p0_failed"): + # Wait until verify step exists before failing on p0 alone + pass + print( + f"attempt {attempt}: status={status} steps={list(steps.keys())} " + f"releasable={releasable}" + ) + time.sleep(poll) + print(f"::error::verify timeout after {timeout_sec}s for {tid}", file=sys.stderr) + return 1 + + def main() -> int: if len(sys.argv) < 2: - print("usage: qa_agent_notify_ci.py detect|wait|notify", file=sys.stderr) + print("usage: qa_agent_notify_ci.py detect|wait|notify|wait-verify", file=sys.stderr) return 2 cmd = sys.argv[1] if cmd == "detect": @@ -183,6 +290,8 @@ def main() -> int: return cmd_wait() if cmd == "notify": return cmd_notify() + if cmd in {"wait-verify", "wait_verify"}: + return cmd_wait_verify() print(f"unknown command: {cmd}", file=sys.stderr) return 2 diff --git a/services/am-analysis/src/main/java/com/am/analysis/service/bootstrap/TriggerCalculationPublisher.java b/services/am-analysis/src/main/java/com/am/analysis/service/bootstrap/TriggerCalculationPublisher.java index 73b160818..41511f82a 100644 --- a/services/am-analysis/src/main/java/com/am/analysis/service/bootstrap/TriggerCalculationPublisher.java +++ b/services/am-analysis/src/main/java/com/am/analysis/service/bootstrap/TriggerCalculationPublisher.java @@ -8,15 +8,21 @@ import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.stereotype.Service; import java.time.Instant; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; @Service @RequiredArgsConstructor +@Slf4j public class TriggerCalculationPublisher { + private static final long PUBLISH_TIMEOUT_SECONDS = 2L; + private final KafkaTemplate kafkaTemplate; private final ObjectMapper objectMapper; private final FlowLogger flowLogger; @@ -47,13 +53,23 @@ public void publish(String userId, String portfolioId, String source, String inh String payload = objectMapper.writeValueAsString(event); String key = portfolioId != null ? portfolioId : (userId != null ? userId : "global"); - kafkaTemplate.send(KafkaTopics.TRIGGER_CALCULATION, key, payload); + // Never block HTTP read path for ~60s when Kafka metadata is unavailable. + kafkaTemplate.send(KafkaTopics.TRIGGER_CALCULATION, key, payload) + .get(PUBLISH_TIMEOUT_SECONDS, TimeUnit.SECONDS); flowLogger.complete(span, "payload_bytes", payload.length(), "trace_id_used", traceId); + } catch (TimeoutException e) { + log.warn("[Bootstrap] Kafka trigger publish timed out ({}s) for portfolioId={} — returning without blocking HTTP", + PUBLISH_TIMEOUT_SECONDS, debounceKey); + flowLogger.fail(span, e); } catch (JsonProcessingException e) { flowLogger.fail(span, e); + } catch (Exception e) { + log.warn("[Bootstrap] Kafka trigger publish failed for portfolioId={}: {}", + debounceKey, e.getMessage()); + flowLogger.fail(span, e); } } } diff --git a/services/am-analysis/src/main/resources/application.yml b/services/am-analysis/src/main/resources/application.yml index 095f965ff..39b54c640 100644 --- a/services/am-analysis/src/main/resources/application.yml +++ b/services/am-analysis/src/main/resources/application.yml @@ -17,6 +17,10 @@ spring: auto-offset-reset: latest key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.apache.kafka.common.serialization.StringDeserializer + producer: + # Cap metadata/buffer block so HTTP bootstrap triggers cannot stall ~60s when Kafka is down. + properties: + max.block.ms: ${KAFKA_PRODUCER_MAX_BLOCK_MS:2000} properties: security-protocol: ${KAFKA_SECURITY_PROTOCOL:SASL_PLAINTEXT} sasl-mechanism: ${KAFKA_SASL_MECHANISM:SCRAM-SHA-256} diff --git a/services/am-mcp-server/pom.xml b/services/am-mcp-server/pom.xml index f62ae42ac..94ec47f2a 100644 --- a/services/am-mcp-server/pom.xml +++ b/services/am-mcp-server/pom.xml @@ -1,4 +1,5 @@ + - - 1.1.4-SNAPSHOT - 1.1.4-SNAPSHOT - 1.1.4-SNAPSHOT + + 1.1.5-SNAPSHOT + 1.1.5-SNAPSHOT + 1.1.5-SNAPSHOT 1.1.4-SNAPSHOT - 1.1.4-SNAPSHOT - 1.1.4-SNAPSHOT - 1.1.4-SNAPSHOT - 1.1.4-SNAPSHOT - 1.1.4-SNAPSHOT + 1.1.5-SNAPSHOT + 1.1.5-SNAPSHOT + 1.0.0-SNAPSHOT + 1.1.3-SNAPSHOT + 1.1.3-SNAPSHOT 1.0.0-M6 2.2.0 diff --git a/services/am-mcp-server/scripts/mcp_remote_launch.py b/services/am-mcp-server/scripts/mcp_remote_launch.py index feac3ab1b..64232f6b6 100644 --- a/services/am-mcp-server/scripts/mcp_remote_launch.py +++ b/services/am-mcp-server/scripts/mcp_remote_launch.py @@ -4,15 +4,16 @@ Bridges local stdio <-> remote HTTP/SSE so ChatGPT, Claude, Cursor, and other LLM tools can attach to am-mcp-server the same way. +Auth (prefer Keycloak client_credentials — same as platform agents): + AM_MCP_CLIENT_ID / AM_MCP_CLIENT_SECRET + KEYCLOAK_TOKEN_URL +Fallback (human / break-glass): + AM_AUTH_USER / AM_AUTH_PASS → AM_AUTH_LOGIN_URL (am-identity) + Env: - AM_MCP_SSE_URL default http://127.0.0.1:18080/sse (kubectl port-forward) - public: https://am-dev.asrax.in/mcp/sse - AM_AUTH_LOGIN_URL default https://am-dev.asrax.in/identity/auth/login - (falls back to identity port-forward if public login fails) - AM_AUTH_USER / AM_AUTH_PASS identity credentials (Bearer for /sse) - AM_MCP_PF_PORT local port for am-mcp-server forward (default 18080) - AM_IDENTITY_PF_PORT local port for am-identity forward (default 18001) - KUBECONFIG required for auto port-forward when using localhost SSE + AM_MCP_SSE_URL default https://am-dev.asrax.in/mcp/sse + KEYCLOAK_TOKEN_URL default http://auth.munish.org/auth/realms/am-dev-realm/protocol/openid-connect/token + AM_AUTH_LOGIN_URL default https://am-dev.asrax.in/identity/auth/login + AM_MCP_PF_PORT / AM_IDENTITY_PF_PORT / KUBECONFIG — only for localhost SSE """ from __future__ import annotations @@ -25,6 +26,7 @@ import sys import time import urllib.error +import urllib.parse import urllib.request @@ -71,6 +73,32 @@ def _ensure_port_forward(local_port: int, service: str) -> None: raise RuntimeError(f"port-forward to 127.0.0.1:{local_port} ({service}) did not come up") +def _client_credentials(token_url: str, client_id: str, client_secret: str) -> str: + body = urllib.parse.urlencode( + { + "grant_type": "client_credentials", + "client_id": client_id, + "client_secret": client_secret, + } + ).encode() + req = urllib.request.Request( + token_url, + data=body, + headers={ + "Content-Type": "application/x-www-form-urlencoded", + "Accept": "application/json", + "User-Agent": "am-mcp-remote-launch/1.1", + }, + method="POST", + ) + with urllib.request.urlopen(req, timeout=30) as resp: + data = json.loads(resp.read().decode()) + token = data.get("access_token") or "" + if not token: + raise RuntimeError(f"client_credentials ok but no access_token: keys={list(data)}") + return token + + def _login(url: str, user: str, password: str) -> str: body = json.dumps({"username": user, "password": password}).encode() req = urllib.request.Request( @@ -79,7 +107,7 @@ def _login(url: str, user: str, password: str) -> str: headers={ "Content-Type": "application/json", "Accept": "application/json", - "User-Agent": "am-mcp-remote-launch/1.0", + "User-Agent": "am-mcp-remote-launch/1.1", }, method="POST", ) @@ -104,18 +132,48 @@ def _login_with_fallback(login_url: str, user: str, password: str, identity_port return _login(fallback, user, password) +def _resolve_token( + *, + token_url: str, + client_id: str, + client_secret: str, + login_url: str, + user: str, + password: str, + identity_port: int, +) -> str: + if client_id and client_secret: + try: + print("auth: Keycloak client_credentials", file=sys.stderr) + return _client_credentials(token_url, client_id, client_secret) + except Exception as exc: # noqa: BLE001 + print(f"client_credentials failed: {exc}", file=sys.stderr) + if not (user and password): + raise + print("auth: falling back to identity user login", file=sys.stderr) + if user and password: + return _login_with_fallback(login_url, user, password, identity_port) + raise RuntimeError( + "Set AM_MCP_CLIENT_ID+AM_MCP_CLIENT_SECRET (preferred) " + "or AM_AUTH_USER+AM_AUTH_PASS" + ) + + def main() -> int: local_port = int(os.environ.get("AM_MCP_PF_PORT", "18080")) identity_port = int(os.environ.get("AM_IDENTITY_PF_PORT", "18001")) - sse_url = os.environ.get("AM_MCP_SSE_URL", f"http://127.0.0.1:{local_port}/sse") + sse_url = os.environ.get("AM_MCP_SSE_URL", "https://am-dev.asrax.in/mcp/sse") + token_url = os.environ.get( + "KEYCLOAK_TOKEN_URL", + "http://auth.munish.org/auth/realms/am-dev-realm/protocol/openid-connect/token", + ) login_url = os.environ.get( "AM_AUTH_LOGIN_URL", "https://am-dev.asrax.in/identity/auth/login" ) - user = os.environ.get("AM_AUTH_USER", "") - password = os.environ.get("AM_AUTH_PASS", "") - if not user or not password: - print("AM_AUTH_USER and AM_AUTH_PASS are required", file=sys.stderr) - return 2 + client_id = os.environ.get("AM_MCP_CLIENT_ID", "").strip() + client_secret = os.environ.get("AM_MCP_CLIENT_SECRET", "").strip() + user = os.environ.get("AM_AUTH_USER", "").strip() + password = os.environ.get("AM_AUTH_PASS", "").strip() if sse_url.startswith("http://127.0.0.1") or sse_url.startswith("http://localhost"): try: @@ -125,9 +183,17 @@ def main() -> int: return 1 try: - token = _login_with_fallback(login_url, user, password, identity_port) + token = _resolve_token( + token_url=token_url, + client_id=client_id, + client_secret=client_secret, + login_url=login_url, + user=user, + password=password, + identity_port=identity_port, + ) except Exception as exc: # noqa: BLE001 - print(f"identity login failed: {exc}", file=sys.stderr) + print(f"auth failed: {exc}", file=sys.stderr) return 1 npx = shutil.which("npx.cmd") or shutil.which("npx") diff --git a/services/am-mcp-server/src/main/java/com/am/mcp/auth/AuthTokenProvider.java b/services/am-mcp-server/src/main/java/com/am/mcp/auth/AuthTokenProvider.java index f9b9434c3..ea1a1c764 100644 --- a/services/am-mcp-server/src/main/java/com/am/mcp/auth/AuthTokenProvider.java +++ b/services/am-mcp-server/src/main/java/com/am/mcp/auth/AuthTokenProvider.java @@ -1,6 +1,7 @@ package com.am.mcp.auth; import com.am.mcp.config.AmMcpProperties; +import com.am.security.context.UserContext; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.http.HttpHeaders; @@ -14,13 +15,11 @@ import java.util.concurrent.locks.ReentrantLock; /** - * Thread-safe JWT provider (AM Platform identity). Priority: - * 1. am.auth.static-token (set in application.yaml) → use directly, no network - * 2. Valid cached JWT (TTL not expired) → return cached - * 3. POST {am.auth.url}/auth/login → fetch, cache, return - * - * Same contract as SPT / other platform services: am-identity POST /auth/login - * body {username,password} → {access_token, ...}. + * JWT provider for outbound calls to portfolio / market / trade / AI. + * Priority: + * 1. Inbound request Bearer ({@link UserContext#getToken()}) — user identity + * 2. {@code am.auth.static-token} — ops override + * 3. Cached service login via am-identity — fallback when no user context */ @Slf4j @Component @@ -28,18 +27,26 @@ public class AuthTokenProvider { private final AmMcpProperties props; - private final RestClient restClient; + private final RestClient restClient; - private volatile String cachedToken = null; - private volatile Instant tokenExpiry = Instant.EPOCH; - private final ReentrantLock lock = new ReentrantLock(); + private volatile String cachedToken = null; + private volatile Instant tokenExpiry = Instant.EPOCH; + private final ReentrantLock lock = new ReentrantLock(); public String getToken() { + String requestToken = UserContext.getToken(); + if (requestToken != null && !requestToken.isBlank()) { + return requestToken; + } + String staticToken = props.getAuth().getStaticToken(); if (staticToken != null && !staticToken.isBlank()) { return staticToken; } - if (isValid()) return cachedToken; + + if (isValid()) { + return cachedToken; + } return fetch(); } @@ -47,12 +54,12 @@ public String getToken() { public HttpHeaders authHeaders() { HttpHeaders h = new HttpHeaders(); String token = getToken(); - if (token != null && !token.isBlank()) h.setBearerAuth(token); + if (token != null && !token.isBlank()) { + h.setBearerAuth(token); + } return h; } - // ── Private ─────────────────────────────────────────────────────────────── - private boolean isValid() { return cachedToken != null && Instant.now().isBefore( @@ -63,7 +70,9 @@ private boolean isValid() { private String fetch() { lock.lock(); try { - if (isValid()) return cachedToken; // double-check after lock + if (isValid()) { + return cachedToken; + } String username = props.getAuth().getUsername(); String password = props.getAuth().getPassword(); @@ -82,18 +91,21 @@ private String fetch() { .retrieve() .body(Map.class); - if (data == null) throw new IllegalStateException("Empty auth response"); + if (data == null) { + throw new IllegalStateException("Empty auth response"); + } String token = (String) data.getOrDefault("access_token", - data.getOrDefault("accessToken", - data.getOrDefault("token", ""))); - if (token == null || token.isBlank()) + data.getOrDefault("accessToken", + data.getOrDefault("token", ""))); + if (token == null || token.isBlank()) { throw new IllegalStateException("Auth response missing access_token: " + data); + } int ttl = ((Number) data.getOrDefault("expires_in", - data.getOrDefault("expiresIn", 3600))).intValue(); - cachedToken = token; - tokenExpiry = Instant.now().plusSeconds(ttl); + data.getOrDefault("expiresIn", 3600))).intValue(); + cachedToken = token; + tokenExpiry = Instant.now().plusSeconds(ttl); log.info("JWT acquired from am-identity (TTL={}s)", ttl); return token; @@ -107,4 +119,4 @@ private String fetch() { lock.unlock(); } } -} \ No newline at end of file +} diff --git a/services/am-mcp-server/src/main/java/com/am/mcp/config/TradeSdkConfig.java b/services/am-mcp-server/src/main/java/com/am/mcp/config/TradeSdkConfig.java index f11238907..73c320e77 100644 --- a/services/am-mcp-server/src/main/java/com/am/mcp/config/TradeSdkConfig.java +++ b/services/am-mcp-server/src/main/java/com/am/mcp/config/TradeSdkConfig.java @@ -2,13 +2,15 @@ import am.trade.sdk.AmTradeSdk; import com.am.mcp.auth.AuthTokenProvider; +import com.am.trade.client.auth.TradeAuthTokenSupplier; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Primary; /** - * Trade SDK with live JWT from AuthTokenProvider (overrides library default bean). + * Wires AmTradeSdk with blank apiKey at init; outbound trade calls use the live + * per-request user JWT from {@link AuthTokenProvider} via {@link TradeAuthTokenSupplier}. */ @Configuration public class TradeSdkConfig { @@ -16,12 +18,15 @@ public class TradeSdkConfig { @Bean @Primary public AmTradeSdk amTradeSdk( - @Value("${am.services.trade-url:http://localhost:8040}") String tradeUrl, - AuthTokenProvider authTokenProvider) { - String token = authTokenProvider.getToken(); + @Value("${am.services.trade-url:http://localhost:8040}") String tradeUrl) { return AmTradeSdk.builder() .apiUrl(tradeUrl) - .apiKey(token != null ? token : "") + .apiKey("") .build(); } + + @Bean + public TradeAuthTokenSupplier tradeAuthTokenSupplier(AuthTokenProvider authTokenProvider) { + return authTokenProvider::getToken; + } } diff --git a/services/am-mcp-server/src/main/java/com/am/mcp/tools/AiAgentTools.java b/services/am-mcp-server/src/main/java/com/am/mcp/tools/AiAgentTools.java index 522927206..d401b66ad 100644 --- a/services/am-mcp-server/src/main/java/com/am/mcp/tools/AiAgentTools.java +++ b/services/am-mcp-server/src/main/java/com/am/mcp/tools/AiAgentTools.java @@ -84,6 +84,6 @@ public String agentFallback(String u, String q, String s, Exception e) { } private String resolve(String userId) { - return (userId != null && !userId.isBlank()) ? userId : props.getDefaults().getUserId(); + return com.am.mcp.util.UserIdResolver.resolve(userId, props); } } diff --git a/services/am-mcp-server/src/main/java/com/am/mcp/tools/AnalysisTools.java b/services/am-mcp-server/src/main/java/com/am/mcp/tools/AnalysisTools.java index ea2b78827..f93789bc7 100644 --- a/services/am-mcp-server/src/main/java/com/am/mcp/tools/AnalysisTools.java +++ b/services/am-mcp-server/src/main/java/com/am/mcp/tools/AnalysisTools.java @@ -136,6 +136,6 @@ public String marketCapFallback(String u, Exception e) { } private String resolve(String userId) { - return (userId != null && !userId.isBlank()) ? userId : props.getDefaults().getUserId(); + return com.am.mcp.util.UserIdResolver.resolve(userId, props); } } diff --git a/services/am-mcp-server/src/main/java/com/am/mcp/tools/BasketTools.java b/services/am-mcp-server/src/main/java/com/am/mcp/tools/BasketTools.java index 34f93a930..d3ca120ea 100644 --- a/services/am-mcp-server/src/main/java/com/am/mcp/tools/BasketTools.java +++ b/services/am-mcp-server/src/main/java/com/am/mcp/tools/BasketTools.java @@ -154,8 +154,7 @@ public String calculateFallback(Double a, String o, Exception e) { private Map baseBody(String portfolioId, String userId) { Map body = new HashMap<>(); body.put("portfolioId", portfolioId); - body.put("userId", (userId != null && !userId.isBlank()) - ? userId : props.getDefaults().getUserId()); + body.put("userId", com.am.mcp.util.UserIdResolver.resolve(userId, props)); return body; } } diff --git a/services/am-mcp-server/src/main/java/com/am/mcp/tools/PortfolioTools.java b/services/am-mcp-server/src/main/java/com/am/mcp/tools/PortfolioTools.java index a8427070d..fd4bb7483 100644 --- a/services/am-mcp-server/src/main/java/com/am/mcp/tools/PortfolioTools.java +++ b/services/am-mcp-server/src/main/java/com/am/mcp/tools/PortfolioTools.java @@ -24,7 +24,7 @@ /** * Portfolio domain MCP tools. - * Identity comes from the service JWT (AuthTokenProvider → Bearer). + * Identity comes from the inbound user JWT (AuthTokenProvider → Bearer). */ @Slf4j @Service diff --git a/services/am-mcp-server/src/main/java/com/am/mcp/tools/TradeTools.java b/services/am-mcp-server/src/main/java/com/am/mcp/tools/TradeTools.java index 2102b1fac..7d2912f7e 100644 --- a/services/am-mcp-server/src/main/java/com/am/mcp/tools/TradeTools.java +++ b/services/am-mcp-server/src/main/java/com/am/mcp/tools/TradeTools.java @@ -241,6 +241,6 @@ public String tradeSummaryFallback(String p, Exception e) { } private String resolve(String userId) { - return (userId != null && !userId.isBlank()) ? userId : props.getDefaults().getUserId(); + return com.am.mcp.util.UserIdResolver.resolve(userId, props); } } diff --git a/services/am-mcp-server/src/main/java/com/am/mcp/util/UserIdResolver.java b/services/am-mcp-server/src/main/java/com/am/mcp/util/UserIdResolver.java new file mode 100644 index 000000000..52a140e11 --- /dev/null +++ b/services/am-mcp-server/src/main/java/com/am/mcp/util/UserIdResolver.java @@ -0,0 +1,24 @@ +package com.am.mcp.util; + +import com.am.mcp.config.AmMcpProperties; +import com.am.security.context.UserContext; + +/** + * Resolves tool userId: explicit arg → inbound JWT sub → configured default. + */ +public final class UserIdResolver { + + private UserIdResolver() { + } + + public static String resolve(String userId, AmMcpProperties props) { + if (userId != null && !userId.isBlank()) { + return userId; + } + String fromJwt = UserContext.getUserId(); + if (fromJwt != null && !fromJwt.isBlank()) { + return fromJwt; + } + return props.getDefaults().getUserId(); + } +} diff --git a/services/am-mcp-server/src/test/java/com/am/mcp/auth/AuthTokenProviderTest.java b/services/am-mcp-server/src/test/java/com/am/mcp/auth/AuthTokenProviderTest.java new file mode 100644 index 000000000..0ccc9e587 --- /dev/null +++ b/services/am-mcp-server/src/test/java/com/am/mcp/auth/AuthTokenProviderTest.java @@ -0,0 +1,42 @@ +package com.am.mcp.auth; + +import com.am.mcp.config.AmMcpProperties; +import com.am.security.context.UserContext; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.springframework.web.client.RestClient; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; + +class AuthTokenProviderTest { + + @AfterEach + void clear() { + UserContext.clear(); + } + + @Test + void prefersInboundUserTokenOverStaticAndServiceLogin() { + AmMcpProperties props = new AmMcpProperties(); + props.getAuth().setStaticToken("static-token"); + props.getAuth().setUsername("svc"); + props.getAuth().setPassword("pw"); + props.getAuth().setUrl("http://identity"); + + AuthTokenProvider provider = new AuthTokenProvider(props, mock(RestClient.class)); + UserContext.setToken("user-jwt"); + + assertThat(provider.getToken()).isEqualTo("user-jwt"); + } + + @Test + void fallsBackToStaticWhenNoUserContext() { + AmMcpProperties props = new AmMcpProperties(); + props.getAuth().setStaticToken("static-token"); + + AuthTokenProvider provider = new AuthTokenProvider(props, mock(RestClient.class)); + + assertThat(provider.getToken()).isEqualTo("static-token"); + } +} diff --git a/services/am-mcp-server/src/test/java/com/am/mcp/util/UserIdResolverTest.java b/services/am-mcp-server/src/test/java/com/am/mcp/util/UserIdResolverTest.java new file mode 100644 index 000000000..567cdc23b --- /dev/null +++ b/services/am-mcp-server/src/test/java/com/am/mcp/util/UserIdResolverTest.java @@ -0,0 +1,43 @@ +package com.am.mcp.util; + +import com.am.mcp.config.AmMcpProperties; +import com.am.security.context.UserContext; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +class UserIdResolverTest { + + @AfterEach + void clear() { + UserContext.clear(); + } + + @Test + void prefersExplicitArg() { + AmMcpProperties props = new AmMcpProperties(); + props.getDefaults().setUserId("default-user"); + UserContext.setUserId("jwt-user"); + + assertThat(UserIdResolver.resolve("arg-user", props)).isEqualTo("arg-user"); + } + + @Test + void usesJwtWhenArgBlank() { + AmMcpProperties props = new AmMcpProperties(); + props.getDefaults().setUserId("default-user"); + UserContext.setUserId("jwt-user"); + + assertThat(UserIdResolver.resolve(" ", props)).isEqualTo("jwt-user"); + assertThat(UserIdResolver.resolve(null, props)).isEqualTo("jwt-user"); + } + + @Test + void usesDefaultWhenNoJwt() { + AmMcpProperties props = new AmMcpProperties(); + props.getDefaults().setUserId("default-user"); + + assertThat(UserIdResolver.resolve(null, props)).isEqualTo("default-user"); + } +}