Merge pull request #48 from oracle/fix/qualified-db-objects

doc/source/user_guide/a2a.rst

@@ -780,6 +780,8 @@ Important clustered options include:
   internal worker infrastructure.
 * ``--worker-replicas``: number of worker runtimes available for new sessions.
 * ``--session-ttl-seconds``: lifetime of temporary database sessions.
+* ``--conversation-retention-days``: number of days to retain Oracle
+  conversations (default 7; use 0 to disable automatic removal).
 * ``--enable-worker-mtls``: create and use server-to-worker certificates.
 * ``--rotate-worker-mtls``: replace the existing test certificates and restart
   the worker workload.

doc/source/user_guide/cli.rst

@@ -288,10 +288,13 @@ passing the A2UI form values to the worker.
         --consul-url http://127.0.0.1:8500
 
 The worker options are ``--host``, ``--port``, ``--worker-id``,
-``--consul-url``, ``--worker-endpoint``, ``--session-ttl-seconds``, and
-``--session-start-timeout-seconds``. ``WORKER_ID``, ``CONSUL_HTTP_URL``, and
-``WORKER_ENDPOINT`` are environment fallbacks; explicit command-line values
-take precedence. ``--port`` is also the port registered with Consul. Cluster
+``--consul-url``, ``--worker-endpoint``, ``--session-ttl-seconds``,
+``--conversation-retention-days``, and ``--session-start-timeout-seconds``.
+``WORKER_ID``, ``CONSUL_HTTP_URL``, ``WORKER_ENDPOINT``, and
+``CONVERSATION_RETENTION_DAYS`` are environment fallbacks; explicit
+command-line values take precedence. Conversation retention defaults to 7 days;
+use 0 to keep conversations until manually deleted. ``--port`` is also the port registered with Consul.
+Cluster
 manifests can set ``WORKER_ADDRESS`` to the pod IP when no endpoint is
 supplied. The worker's ``--tls-cert-file``, ``--tls-key-file``, and
 ``--tls-ca-file`` options enable gateway-to-worker mTLS; provide all three

gcloud/cluster/README.md

@@ -6,6 +6,10 @@ password through the generated A2UI form. A session remains available for 15
 minutes by default. Set a
 different lifetime in seconds with `--session-ttl-seconds`; for example,
 `--session-ttl-seconds 1800` keeps sessions for 30 minutes.
+Oracle conversation history is retained for 7 days by default. Set
+`--conversation-retention-days` to change this, or use 0 to keep
+conversations until they are
+manually deleted.
 
 ## Protocol architecture
 

gcloud/cluster/cloudbuild.yaml

@@ -131,6 +131,7 @@ steps:
         sed -e "s|WORKER_IMAGE|$image|" \
           -e "s|WORKER_REPLICAS|${_WORKER_REPLICAS}|" \
           -e "s|SESSION_TTL_SECONDS|${_SESSION_TTL_SECONDS}|" \
+          -e "s|CONVERSATION_RETENTION_DAYS|${_CONVERSATION_RETENTION_DAYS}|" \
           -e "s|GKE_DNS_DOMAIN|${_GKE_DNS_DOMAIN}|" \
           "$worker_manifest" > /workspace/worker.yaml
 
@@ -222,6 +223,7 @@ substitutions:
   _SUBNET: default
   _WORKER_REPLICAS: "2"
   _SESSION_TTL_SECONDS: "900"
+  _CONVERSATION_RETENTION_DAYS: "7"
   _ENABLE_WORKER_MTLS: "false"
   _ROTATE_WORKER_MTLS: "false"
   _GKE_DNS_DOMAIN: select-ai-a2a-cluster.internal

gcloud/cluster/deploy.sh

@@ -34,6 +34,9 @@ Options:
   --subnet NAME               VPC subnet for Cloud Run direct VPC egress (default: default)
   --worker-replicas COUNT     GKE worker replica count (default: 2)
   --session-ttl-seconds N    Dynamic session lifetime (default: 900)
+  --conversation-retention-days N
+                             Oracle conversation retention (default: 7; 0 keeps
+                             conversations until manually deleted)
   --enable-worker-mtls        Use ephemeral mTLS certificates for server-to-worker calls
   --rotate-worker-mtls        Replace the existing worker mTLS CA and certificates
   --mtls-cert-validity-days DAYS
@@ -59,6 +62,7 @@ network="default"
 subnet="default"
 worker_replicas="2"
 session_ttl_seconds="900"
+conversation_retention_days="7"
 enable_worker_mtls="false"
 rotate_worker_mtls="false"
 mtls_cert_validity_days="365"
@@ -84,6 +88,10 @@ while [[ $# -gt 0 ]]; do
       session_ttl_seconds="${2:?--session-ttl-seconds requires a value}"
       shift 2
       ;;
+    --conversation-retention-days)
+      conversation_retention_days="${2:?--conversation-retention-days requires a value}"
+      shift 2
+      ;;
     --enable-worker-mtls) enable_worker_mtls="true"; shift ;;
     --rotate-worker-mtls) rotate_worker_mtls="true"; shift ;;
     --mtls-cert-validity-days)
@@ -113,6 +121,10 @@ if ! [[ "$session_ttl_seconds" =~ ^[1-9][0-9]*$ ]]; then
   echo "--session-ttl-seconds must be a positive integer." >&2
   exit 2
 fi
+if ! [[ "$conversation_retention_days" =~ ^[0-9]+$ ]]; then
+  echo "--conversation-retention-days must be a non-negative integer." >&2
+  exit 2
+fi
 if [[ -z "$gke_dns_domain" || "$gke_dns_domain" == *.local ]]; then
   echo "--gke-dns-domain must be non-empty and must not end in .local." >&2
   exit 2
@@ -300,6 +312,7 @@ build_substitutions=(
   "_SUBNET=$subnet"
   "_WORKER_REPLICAS=$worker_replicas"
   "_SESSION_TTL_SECONDS=$session_ttl_seconds"
+  "_CONVERSATION_RETENTION_DAYS=$conversation_retention_days"
   "_ENABLE_WORKER_MTLS=$enable_worker_mtls"
   "_ROTATE_WORKER_MTLS=$rotate_worker_mtls"
   "_GKE_DNS_DOMAIN=$gke_dns_domain"

gcloud/cluster/gke/worker-mtls.yaml

@@ -34,6 +34,8 @@ spec:
             - https://$(POD_NAME).select-ai-a2a-worker.select-ai-a2a.svc.GKE_DNS_DOMAIN:8443
             - --session-ttl-seconds
             - "SESSION_TTL_SECONDS"
+            - --conversation-retention-days
+            - "CONVERSATION_RETENTION_DAYS"
             - --tls-cert-file
             - /var/run/select-ai-mtls/tls.crt
             - --tls-key-file

gcloud/cluster/gke/worker.yaml

@@ -34,6 +34,8 @@ spec:
             - http://$(WORKER_ADDRESS):8080
             - --session-ttl-seconds
             - "SESSION_TTL_SECONDS"
+            - --conversation-retention-days
+            - "CONVERSATION_RETENTION_DAYS"
           ports:
             - name: http
               containerPort: 8080

src/select_ai/agent/a2a/context_store.py

@@ -7,6 +7,7 @@
 
 """Oracle Database storage for A2A-to-Oracle conversation mappings."""
 
+import datetime
 from asyncio import Lock
 from typing import Optional
 
@@ -52,8 +53,12 @@ class OracleContextStore:
     def __init__(
         self,
         owner_resolver: OwnerResolver = resolve_user_scope,
+        retention_days: int = 7,
     ) -> None:
+        if retention_days < 0:
+            raise ValueError("retention_days must be non-negative")
         self.owner_resolver = owner_resolver
+        self.retention_days = retention_days
         self.initialized = False
         self.initialize_lock = Lock()
 
@@ -85,6 +90,7 @@ async def get_or_create(
             attributes=select_ai.ConversationAttributes(
                 title=f"A2A {team_name}",
                 description=f"A2A context {context_id}",
+                retention_days=datetime.timedelta(days=self.retention_days),
             )
         )
         conversation_id = await conversation.create()

src/select_ai/agent/a2a/models.py

@@ -80,6 +80,7 @@ class WorkerSettings:
     session_ttl_seconds: int
     session_start_timeout_seconds: int
     worker_endpoint: str | None = None
+    conversation_retention_days: int = 7
 
     def __post_init__(self) -> None:
         if self.worker_port < 1 or self.worker_port > 65_535:
@@ -90,6 +91,10 @@ def __post_init__(self) -> None:
             raise ValueError(
                 "session_start_timeout_seconds must be at least 1"
             )
+        if self.conversation_retention_days < 0:
+            raise ValueError(
+                "conversation_retention_days must be non-negative"
+            )
         object.__setattr__(self, "consul_url", self.consul_url.rstrip("/"))
         if self.worker_endpoint:
             object.__setattr__(

src/select_ai/agent/a2a/session_process.py

@@ -101,9 +101,11 @@ def __init__(
         self,
         session_ttl_seconds: int,
         session_start_timeout_seconds: int,
+        conversation_retention_days: int = 7,
     ) -> None:
         self.session_ttl_seconds = session_ttl_seconds
         self.session_start_timeout_seconds = session_start_timeout_seconds
+        self.conversation_retention_days = conversation_retention_days
         self.sessions: dict[str, ChildSession] = {}
 
     async def open(self, spec: SessionSpec) -> None:
@@ -122,6 +124,7 @@ async def open(self, spec: SessionSpec) -> None:
                 spec.session_id,
                 spec.owner,
                 spec.team_name,
+                self.conversation_retention_days,
             ),
             daemon=True,
         )
@@ -322,6 +325,7 @@ def _session_process_main(
     session_id: str,
     owner: str,
     team_name: str,
+    conversation_retention_days: int,
 ) -> None:
     """Entrypoint for a child that owns one Select AI database session."""
     try:
@@ -332,6 +336,7 @@ def _session_process_main(
                 session_id,
                 owner,
                 team_name,
+                conversation_retention_days,
             )
         )
     finally:
@@ -344,6 +349,7 @@ async def _run_session_process(
     session_id: str,
     owner: str,
     team_name: str,
+    conversation_retention_days: int = 7,
 ) -> None:
     """Open one async connection and execute A2A operations."""
     import select_ai
@@ -358,7 +364,12 @@ async def _run_session_process(
         )
         if not await select_ai.async_is_connected():
             raise RuntimeError("Database login failed.")
-        runtime = SessionRuntime(session_id, owner, team_name)
+        runtime = SessionRuntime(
+            session_id,
+            owner,
+            team_name,
+            conversation_retention_days,
+        )
         await runtime.initialize()
         connection.send({"type": PipeMessageType.READY.value})
         ready = True

src/select_ai/agent/a2a/session_runtime.py

@@ -58,12 +58,20 @@ def user_name(self) -> str:
 class SessionRuntime:
     """Own the A2A handler and Oracle stores for one connected database."""
 
-    def __init__(self, session_id: str, owner: str, team_name: str) -> None:
+    def __init__(
+        self,
+        session_id: str,
+        owner: str,
+        team_name: str,
+        conversation_retention_days: int = 7,
+    ) -> None:
         self.session_id = session_id
         self.owner = owner
         self.team_name = team_name
         self.task_store = OracleTaskStore()
-        self.context_store = OracleContextStore()
+        self.context_store = OracleContextStore(
+            retention_days=conversation_retention_days,
+        )
         self.handler: DefaultRequestHandler | None = None
 
     async def initialize(self) -> None:

src/select_ai/agent/a2a/worker.py

@@ -130,6 +130,7 @@ def create_worker_app(settings: WorkerSettings) -> FastAPI:
     backend = ProcessSessionBackend(
         settings.session_ttl_seconds,
         settings.session_start_timeout_seconds,
+        settings.conversation_retention_days,
     )
 
     @asynccontextmanager